canal原理及使用

浩客4个月前技术文章176

什么是canal

canal,主要用途是基于 MySQL 数据库增量日志解析,提供增量数据订阅和消费

这里我们可以简单地把canal理解为一个用来同步增量数据的一个工具

Dingtalk_20231229030412.jpg

工作原理

MySQL主备复制原理

  • MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events,可以通过 show binlog events 进行查看)

  • MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)

  • MySQL slave 重放 relay log 中事件,将数据变更反映它自己的数据

canal 工作原理

  • canal 模拟 MySQL slave 的交互协议,伪装自己为 MySQL slave ,向 MySQL master 发送dump 协议

  • MySQL master 收到 dump 请求,开始推送 binary log 给 slave (即 canal )

  • canal 解析 binary log 对象(原始为 byte 流)

1、开启mysql binlog功能
开启之后对mysql的性能会有一定的影响

# 修改my.cnf
vim /etc/my.cnf

[mysqld]    
# 打开binlog
log-bin=mysql-bin
# 选择ROW(行)模式
binlog-format=ROW
# 配置MySQL replaction需要定义,不要和canal的slaveId重复
server_id=1


# 改了配置文件之后,重启MySQL,使用命令查看是否打开binlog模式:
service mysqld restart
# 进入mysql命令行验证一下
show variables like 'log_bin';


2、搭建canal
# 解压到指定目录下
tar -xvf canal.deployer-1.1.4.tar.gz 

#修改配置文件conf/example/instance.properties
# mysql地址
canal.instance.master.address=master:3306
# mysql用户名和密码
canal.instance.dbUsername=root
canal.instance.dbPassword=123456

# 增加canal数据采集之后写入kafkatopic
# 动态topic,每一个表自动创建一个topic
canal.mq.dynamicTopic=bigdata\\..*

# 修改配置文件conf/canal.properties
# zookeeper的地址,如果有多台需要写多个使用逗号分隔
canal.zkServers = master:2181
# canal保存数据的位置
canal.serverMode = kafka
# kafka地址列表
canal.mq.servers = master:9092

# 启动Canal   启动后会出现进程CanalLauncher
# 启动canal之前需要先保证kafka处于启动状态
cd /usr/local/soft/canal/bin/
# 启动
./startup.sh
# 停止
stop.sh
# 重启
restart.sh


## 测试

-- 在mysql创建表插入数据

mysql -uroot -p123456

use `bigdata`;

-- 创建表
CREATE TABLE `order`
(
    id          BIGINT UNIQUE PRIMARY KEY AUTO_INCREMENT COMMENT '主键',
    order_id    VARCHAR(64)   COMMENT '订单ID',
    amount      DECIMAL(10, 2) COMMENT '订单金额',
    create_time DATETIME       COMMENT '创建时间',
    UNIQUE uniq_order_id (`order_id`)
) COMMENT '订单表';

--  插入数据

INSERT INTO `order`(order_id, amount) VALUES ('100', 66);
UPDATE `order` SET amount = 88 WHERE order_id = '100';
DELETE  FROM `order` WHERE order_id = '100';

-- 查看kafka中是否出现topic

kafka-topics.sh --list  --zookeeper  master:2181,node1:2181,node2:2181

--会出现bigdata.order

-- 消费数据
kafka-console-consumer.sh --bootstrap-server master:9092 --from-beginning --topic bigdata.order
返回列表

上一篇:数据湖Iceberg

下一篇:FLink-Canal

相关文章

企业级大数据安全架构(六)

企业级大数据安全架构(六)

本节详细介绍企业级大数据架构中的第六部分,数据授权和审计管理1.Ranger简介Apache Ranger是一款被设计成全面掌管Hadoop生态系统的数据安全管理框架,为Hadoop生态系统...

EMR yarn集成Cgroup

EMR yarn集成Cgroup

CGroups(Control Groups)是Linux内核的一个资源隔离功能,限制/隔离/统计进程的资源使用,包括CPU/内存/磁盘IO/网络等资源。YARN中也集成了CGroups的功能,使得N...

ip_foward未开启导致slb健康检查异常

ip_foward未开启导致slb健康检查异常

现象:在slb上配置一条7层http监听,并开启健康检查,发现健康检查一直异常。处理步骤:针对slb健康检查异常的问题,可从以下四个方面入手,排查原因。1、确认服务器上业务程序正常启动,且在后端服务器...

mysqldump导入备份文件报错记录

mysqldump导入备份文件报错记录

mysqldump导入备份文件报错记录【问题记录1】目前遇到过两类报错,均主要是因为 GTID 引起,两类报错内容完全不同,但解决方法相同,报错内容具体如下:第一类报错:在自建数据库往云上数据库导入数...

LINUX 安全运维-用户密码

密码策略linux作为一个多用户的系统,我们还是不可避免的会去新增很多用户,我们不能保证每一个用户具有很好的安全意识,所以只能在用户的密码以及用户的远程访问上做一些限制,我们先介绍Linux用户密码策...

CPU--平均负载

1、原理概述平均负载是指单位时间内,系统处于可运行状态和不可中断状态的平均进程数,也就是平均活跃进程数,它和 CPU 使用率并没有直接关系。   * ...

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。