简单实现MySQL数据实时增量同步到Kafka————Maxwell

栗筝i

发布于 2022-12-01 08:48:37

1.3K0

发布于 2022-12-01 08:48:37

文章被收录于专栏：迁移内容

任务需求：将MySQL里的数据实时增量同步到Kafka

1、准备工作

1.1、MySQL方面：开启BinLog

1.1.1、修改my.cnf文件

vi /etc/my.cnf
[mysqld]
server-id  = 1
binlog_format = ROW

1.1.2、重启MySQL,然后登陆到MySQL之后，查看是否已经修改过来:

mysql> show variables like 'binlog_format';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW   |
+---------------+-------+

1.1.3、创建Maxwell用户，并赋予 maxwell 库的一些权限

CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%';

1.2、Kafka准备工作 1.2.1、启动Zookeeper

zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties

1.2.2、启动kafka

kafka-server-start /usr/local/etc/kafka/server.properties

2、Maxwell

2.1、下载安装包 https://github.com/zendesk/maxwell/releases/download/v1.20.0/maxwell-1.20.0.tar.gz 2.2、解压到指定位置

tar -zxvf maxwell-1.20.0.tar.gz

2.3、在MYSQL中创建测试用表（前提你要进入一个库）

CREATE TABLE `test` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `age` int(11) DEFAULT NULL,
  `name` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

2.4、打开Maxwell（要在maxwell安装目录下）

bin/maxwell --user=maxwell --password=123456 --host='127.0.0.1' --producer=stdout

2.5、对数据进行操作

insert into test values(1,22,"小明");
update test set name='whirly' where id=1;
delete from test where id=1;

可以看到Maxwell控制台的输出，测试成功！

{"database":"test","table":"test","type":"insert","ts":1552153502,"xid":832,"commit":true,"data":{"id":1,"age":22,"name":"小明锋"}}
{"database":"test","table":"test","type":"update","ts":1552153502,"xid":833,"commit":true,"data":{"id":1,"age":22,"name":"whirly"},"old":{"name":"小明"}}
{"database":"test","table":"test","type":"delete","ts":1552153502,"xid":834,"commit":true,"data":{"id":1,"age":22,"name":"whirly"}}

3、实现MySQL数据实时增量同步到Kafka

3.1、开启指定到Kafka的MaxWell

bin/maxwell --user='maxwell' --password='123456' --host='127.0.0.1' \
   --producer=kafka --kafka.bootstrap.servers=localhost:9092 --kafka_topic=maxwell --kafka_version=0.10.2.1

3.2、对数据库进行操作

insert into test values(1,22,"小明");
update test set name='whirly' where id=1;
delete from test where id=1;

3.3、启动一个消费者来消费 maxwell topic的消息，观察其输出；

kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic test --from-beginning

3.4、再次执行数据库结果观察，仍然可以得到相同的输出

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2020-06-23，如有侵权请联系 cloudcommunity@tencent.com 删除

kafka

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度