实验环境
本地虚拟机
maraidb 10.8.8
kafka 2.12-3.3.1
maxwell由容器部署
1 mariadb
1.1 配置log_bin
配置文件中加入如下内容
server-id = 111
log_bin = mysql-bin
binlog_format = ROW
expire_logs_days = 1
重启服务
systemctl restart mariadb
查询命令
SHOW VARIABLES LIKE 'log_bin%';
1.2 创建用户
CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%';
flush privileges;
2 kafka
参考前期博文《单节点kafka部署笔记》
2.1 修改配置
修改kafka目录下的config/kraft/server.properties
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://172.17.0.1:9092
2.2 启动kafka
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties
bin/kafka-server-start.sh config/kraft/server.properties &
2.3 创建topic
bin/kafka-topics.sh --create --topic maxwell-mysql --bootstrap-server localhost:9092
3 maxwell
3.1 拉取镜像
docker pull zendesk/maxwell
3.2 测试
创建容器
docker run -it --rm zendesk/maxwell bin/maxwell --user='maxwell' --password='123456' --host='172.17.0.1' --producer=stdout
写入数据
create database company;
CREATE TABLE products (id int(10), name varchar(255), price int(20));
insert into products values (1, "car001", 10000);
即可看到数据库操作
{"database":"company","table":"products","type":"insert","ts":1687524147,"xid":1640,"commit":true,"data":{"id":1,"name":"car001","price":10000}}
中断后容器会自动删除
4 正式使用
4.1 启动容器
docker run -d --name maxwell zendesk/maxwell bin/maxwell --user='maxwell' --password='123456' --host='172.17.0.1' --producer=kafka --kafka.bootstrap.servers='172.17.0.1:9092' --kafka_topic=maxwell --log_level=debug
4.2 kafka读取
创建一个consumer
bin/kafka-console-consumer.sh --topic maxwell --from-beginning --bootstrap-server localhost:9092
即可在终端看到变化数据