canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步
搭建环境
操作系统: CentOS release 6.5 (Final)
MySQL版本: 8.0
JDK版本:11
canal版本: 1.1.4
zookeeper 3.6.2
kafka
mysql 配置
进入MySQL 查看是否开启binlog
注意一点是MySQL8.x
默认开启binlog
SHOW VARIABLES LIKE '%bin%';
文章图片
image.png
执行下面代码 创建新的用户名
canal 密码
QWqw12!@ 赋予
REPLICATION SLAVE
和
REPLICATION CLIENT
权限
CREATE USER canal IDENTIFIED BY 'QWqw12!@';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'QWqw12!@';
文章图片
image.png
#查看mysql binlog 是否开启 ON 开启OFF 未开启
show variables like 'log_bin';
#查看binlog日志文件列表
show binary logs;
#查看binglog 模式 是否为ROW
show variables like 'binlog_format'
#查看当前正在写入的binlog文件:
show master status
安装zookeeper
Canal
和Kafka
集群都依赖于Zookeeper
做服务协调,为了方便管理,一般会独立部署Zookeeper
服务或者Zookeeper
集群cd usr/local/ 进入要安装软件目录
mkdir zk cd zkmkdir datacd ../
下载安装zk
wget http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.6.2/apache-zookeeper-3.6.2-bin.tar.gz
下载前打开http://mirror.bit.edu.cn/apache/zookeeper查看版本号多少
避免404 错误
文章图片
image.png
解压zk
tar -zxvf apache-zookeeper-3.6.2-bin.tar.gzcd apache-zookeeper-3.6.2-bin/conf
复值zoo_sample.cfg 到zoo.cfg 并打开zoo.cfk
cp zoo_sample.cfg zoo.cfg && vim zoo.cfg修改dataDir 为刚才建的data文件夹
文章图片
image.png 然后启动
Zookeeper
:进入zk bin 目录
./bin/zkServer.sh start
文章图片
image.png
tail -100f /usr/local/zk/apache-zookeeper-3.6.2-bin/logs/zookeeper-root-server-asp.out 查看打印日志tail -fn 10 test.log -f 循环查看 -fn 循环查看实时10行数据
安装kafka cd usr/local/ 进入要安装软件目录
在目录下创建kafka 文件夹以及在kafka 下创建 data
mkdir kafkacd kafkamkdir datawget [http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz](http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz)tar -zxvf kafka_2.13-2.6.0.tgz
由于解压后 kafka_2.13-2.4.0/config/server.properties
配置中对应的
zookeeper.connect=localhost:2181已经符合需要,不必修改,
需要修改日志文件的目录
log.dirs为
/data/kafka/data。然后启动
Kafka`服务:文章图片
image.png
【canal+mysql+zookeeper+kafka|canal+mysql+zookeeper+kafka 实时同步】修改文件下log.dirs路径为刚刚创建的
修改server.properties 配置文件下的
listeners为
listeners=PLAINTEXT://localhost:9092
data 目录
注意目录在kafka_2.13-2.6.0 下
执行 ./bin/kafka-server-start.sh /usr/local/kafka/kafka_2.13-2.6.0/config/server.properties
文章图片
image.png 这样启动一旦退出控制台就会结束
Kafka
进程,可以添加-daemon
参数用于控制Kafka
进程后台不挂断运行./bin/kafka-server-start.sh -daemon /usr/local/kafka/kafka_2.13-2.6.0/config/server.properties
canal 安装
cd usr/local/ 进入要安装软件目录
mkdir canal
cd canal
wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz
建议复制链接到浏览器手动下载,在上传服务器,翻墙不翻墙都特慢
- bin# 运维脚本
- conf# 配置文件
canal_local.properties# canal本地配置,一般不需要动
canal.properties# canal服务配置
logback.xml# logback日志配置
metrics# 度量统计配置
spring# spring-实例配置,主要和binlog位置计算、一些策略配置相关,可以在canal.properties选用其中的任意一个配置文件
example# 实例配置文件夹,一般认为单个数据库对应一个独立的实例配置文件夹
instance.properties# 实例配置,一般指单个数据库的配置
- lib# 服务依赖包
- logs# 日志文件输出目录
canal 配置 在开发和测试环境建议把
logback.xml
的日志级别修改为DEBUG
方便定位问题。这里需要关注canal.properties
和instance.properties
两个配置文件。canal.properties
文件中,需要修改:- 去掉
canal.instance.parser.parallelThreadSize = 16
这个配置项的注释,也就是启用此配置项,和实例解析器的线程数相关,不配置会表现为阻塞或者不进行解析。 -
canal.serverMode
配置项指定为kafka
,可选值有tcp
、kafka
和rocketmq
(master
分支或者最新的的v1.1.5-alpha-1
版本,可以选用rabbitmq
),默认是kafka
。 -
canal.mq.servers
配置需要指定为Kafka
服务或者集群Broker
的地址,这里配置为127.0.0.1:9092
。
instance.properties
一般指一个数据库实例的配置,Canal
架构支持一个Canal
服务实例,处理多个数据库实例的binlog
异步解析。instance.properties
需要修改的配置项主要包括:canal.instance.mysql.slaveId
需要配置一个和Master
节点的服务ID
完全不同的值,这里笔者配置为654321
。
- 配置数据源实例,包括地址、用户、密码和目标数据库:
-
canal.instance.master.address
,这里指定为127.0.0.1:3306
。 -
canal.instance.dbUsername
,这里指定为canal
。 -
canal.instance.dbPassword
,这里指定为QWqw12!@
。 - 新增
canal.instance.defaultDatabaseName
,这里指定为test
(需要在MySQL
中建立一个test
数据库,见前面的流程)。
Kafka 相关配置,这里暂时使用静态 topic和单个partition
-
canal.mq.topic
,这里指定为test
,也就是解析完的binlog
结构化数据会发送到Kafka
的命名为test
的topic
中。
canal.mq.partition
,这里指定为0
。
#进入bin目录下运行
./startup.sh
# 查看服务日志
tail -100f ../logs/canal/canal
# 查看实例日志
-- 一般情况下,关注实例日志即可
tail -100f ../logs/example/example.log
canal 启动报错的话可以看看这个
https://www.jianshu.com/p/a0788a8f3832
对需要同步的数据库进行crud操作后
进入kafka bin 目录 运行
./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --from-beginning --topic asp_be
对数据进行增加和修改后
kafka 就会有日志输出
推荐阅读
- Elasticsearch|Elasticsearch 简介
- 实时|实时 OLAP 系统 Druid
- SRS(简单实时视频服务)|SRS(简单实时视频服务) 笔记(1)- 体验
- 有趣的oython|python疲劳驾驶实时检测项目讲解(附代码)
- 百度爱番番实时CDP建设实践
- Injection|Injection For Xcode11 macOS 10.15 Catalina 亲测可用iOS模拟器UI界面调试实时刷新工具
- TiDB Cloud 上线亚马逊云科技 Marketplace,为全球用户提供云端一栈式实时 HTAP 数据库体验
- 首批+唯一!阿里云实时计算 Flink 版通过信通院大数据产品稳定性测试
- 吉利集团携手DataPipeline两周年,让实时数据管理平台“厚”起来
- OpenMLDB在AKULAKU实时特征计算场景的应用