AI 助理
文档备案控制台
开发者社区 数据库 文章 正文

利用maxwell组件监听mysql之binlog日志进行实时同步数据

2022-05-16 1576
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
RDS AI 助手，专业版
RDS Agent（兼容OpenClaw），2核4GB
RDS MySQL DuckDB 分析主实例，基础系列 4核8GB
简介： 利用maxwell组件监听mysql之binlog日志进行实时同步数据

一 maxwell组件介绍


Maxwell是一个守护进程，它能监听并读取MySQL的binlog，然后解析输出为json，支持将数据输出到Kafka、Kinesis或其他流媒体平台，支持表和库过滤。


注意：对于增删改都有输出，但对于truncate操作，没有输出。


源码地址：https://github.com/zendesk/maxwell


下载地址：https://github.com/zendesk/maxwell/releases/download/v1.21.1/maxwell-1.21.1.tar.gz


示意如下：


mysql> insert into `test`.`maxwell` set id = 1, daemon = 'Stanislaw Lem';
  maxwell: {
    "database": "test",
    "table": "maxwell",
    "type": "insert",
    "ts": 1449786310,
    "xid": 940752,
    "commit": true,
    "data": { "id":1, "daemon": "Stanislaw Lem" }
  }
  mysql> update test.maxwell set daemon = 'firebus!  firebus!' where id = 1;
  maxwell: {
    "database": "test",
    "table": "maxwell",
    "type": "update",
    "ts": 1449786341,
    "xid": 940786,
    "commit": true,
    "data": {"id":1, "daemon": "Firebus!  Firebus!"},
    "old":  {"daemon": "Stanislaw Lem"}
  }


二 设备与组件版本梳理:

1.  linux内核版本(CentOS Linux 7)：(命令:uname -a)


Linux slave1 3.10.0-693.el7.x86_64 #1 SMP Tue Aug 22 21:09:27 UTC 2017 x86_64 x86_64 x86_64 GNU/Linux


2.  mysql版本：(SQL命令：select version(); 或 status)


Server version: 5.6.43-log MySQL Community Server (GPL)


3.  maxwell版本：maxwell-1.21.1


4.  kafka版本：kafka_2.11-2.2.0


5.  zookeeper版本：zookeeper-3.4.5-cdh5.7.0


三 设备介绍

二台虚拟机，分别为salve1（192.168.175.21）和slave2（192.168.175.22），slave1上安装mysql和kafka，slave2上启动maxwell守护进程和zookeeper。


四 简要过程梳理


网络异常，图片无法展示
|


主流程示意图


如上图：这次主要介绍从binlog > maxwell > kafka的过程，而kafka后面的过程，就可以有很多种了，比如：


（1）binlog > maxwell > kafka > spark streaming > hdfs、kudu；

（2）binlog > maxwell > kafka > flume > hdfs;

（3）binlog > maxwell > kafka > es > kibana;


第一种spark streaming+hdfs、kudu，是目前我所在公司中使用的场景。简要流程梳理如下:


1.  在slave1上安装mysql；
2.  在slave2上启动maxwell，测试是否可以正常读取binlog；
3.  maxwell初步测试ok；
4.  在slave2上安装并启动zk；
5.  在slave1上安装并启动kafka server；
6.  通过kafka producer和consumer测试启动是否成功；
7.  启动maxwell将解析后的json数据发送到kafka；
8.  启动kafka consumer测试数据是否成功发送。


五 详细过程

1 在slave1（192.168.175.21）上安装mysql


详细过程，可参考笔记： https://www.jianshu.com/p/09936d9c7bf2


（1）在创建root账号并设置远程访问之后，接着创建maxwell账号并设置远程访问和权限：


mysql> CREATE USER 'maxwell'@'%' IDENTIFIED BY 'XXXXXX';
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%';
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';


权限:Maxwell需要权限将状态存储在schema_database选项(默认Maxwell)指定的数据库中。


（2）针对maxwell配置mysql： 确保配置了server_id，并打开了基于行的复制。

可参考maxwell快速入门文档:


https://github.com/zendesk/maxwell/blob/master/docs/docs/quickstart.md


$ vi my.cnf
[mysqld]
server_id=1
log-bin=master
binlog_format=row


注意:  binlog_format是一个基于会话的属性需要关闭所有活动连接才能完全转换为基于行的复制。


mysql配置文件my.cnf查找方式:


$ whereis my
my:/etc/my.cnf


配置完成后，要重启mysql服务，方可生效


2 在slave2上启动maxwell，测试是否可以正常读取binlog


（1）在slave2上测试是否可以进行远程访问数据


mysql -h 192.168.175.21 -P 3306 - u root -proot   #root登录成功后，窗口无需关闭，后面还要接着测试用
mysql -h 192.168.175.21 -P 3306 -u maxwell -p111111


（2）下载并解压maxwell


#解压到指定的文件夹
tar xzvf maxwell-1.21.1.tar.gz -C /usr/loca/hadoop/app


（3）命令行启动maxwell，将解析后的日志输出到控制台进行测试


[root@slave2 maxwell-1.21.1]# pwd
/usr/local/hadoop/app/maxwell-1.21.1
[root@slave2 maxwell-1.21.1]# bin/maxwell --user='maxwell' --password='111111' --host='192.168.175.21' --producer=stdout


启动成功后，会展示如下日志内容：


[root@slave2 maxwell-1.21.1]# bin/maxwell --user='maxwell' --password='111111' --host='192.168.175.21' --producer=stdout
Using kafka version: 1.0.0
15:49:04,967 WARN  MaxwellMetrics - Metrics will not be exposed: metricsReportingType not configured.
15:49:08,334 INFO  Maxwell - Maxwell v1.21.1 is booting (StdoutProducer), starting at Position[BinlogPosition[master.000002:984013], lastHeartbeat=1555141484690]
15:49:10,071 INFO  MysqlSavedSchema - Restoring schema id 2 (last modified at Position[BinlogPosition[master.000001:40754], lastHeartbeat=1555112822550])
15:49:12,759 INFO  MysqlSavedSchema - Restoring schema id 1 (last modified at Position[BinlogPosition[master.000001:3816], lastHeartbeat=0])
15:49:13,122 INFO  MysqlSavedSchema - beginning to play deltas...
15:49:13,138 INFO  MysqlSavedSchema - played 1 deltas in 12ms
15:49:13,426 INFO  BinlogConnectorReplicator - Setting initial binlog pos to: master.000002:984013
15:49:13,997 INFO  BinaryLogClient - Connected to 192.168.175.21:3306 at master.000002/984013 (sid:6379, cid:59)
15:49:14,003 INFO  BinlogConnectorLifecycleListener - Binlog connected.


在slave2上root连接mysql的窗口中，执行insert，delete，update操作：


MySQL [mysql]> insert into tb_dept (id,name,description) values(16,'xiaoman','manger');
Query OK, 1 row affected (0.02 sec)
MySQL [mysql]> delete from tb_dept where id = 16;
Query OK, 1 row affected (0.01 sec)
MySQL [mysql]> update tb_dept set name='xiaofei' where id=14;
Query OK, 1 row affected (0.02 sec)
Rows matched: 1  Changed: 1  Warnings: 0


在maxwell的stdout窗口中会产生如下日志：


{"database":"mysql","table":"tb_dept","type":"insert","ts":1555142065,"xid":6349,"commit":true,"data":{"Id":16,"Name":"xiaoman","description":"manger"}}
{"database":"mysql","table":"tb_dept","type":"delete","ts":1555142096,"xid":6361,"commit":true,"data":{"Id":16,"Name":"xiaoman","description":"manger"}}
{"database":"mysql","table":"tb_dept","type":"update","ts":1555142157,"xid":6383,"commit":true,"data":{"Id":14,"Name":"xiaofei","description":"sales"},"old":{"Name":"xiaoming1"}}


3 在slave2上安装并启动zk


具体安装方式，可参考笔记：https://www.jianshu.com/p/10d5a20ab9b7


注意启动完成后，要检查zk是否安装成功


#查看状态
zkServer.sh status


4 在slave1上安装并启动kafka server：


（1）具体启动安装方式，可参考笔记：https://www.jianshu.com/p/3d017bdbfb3c

修改kafka配置文件 $KAFKA_HOME/config/server.properties:


broker.id=1
listeners=PLAINTEXT://slave1:9092
log.dirs=/usr/local/app/tmp/kafka-logs
zookeeper.connect=slave2:2181


（2）通过kafka producer和consumer测试启动是否成功


5 在slave1的kafka上创建名为maxwell的topic


（1）创建topic


kafka-topics.sh --create --zookeeper slave2:2181 --replication-factor 1 --partitions 1 --topic maxwell


（2）检查topic是否创建成功


#查看topic列表：
kafka-topics.sh --list --zookeeper slave2:2181
#查看topic具体描述：
kafka-topics.sh --describe --zookeeper slave2:2181  --topic maxwell


6 在slave1上启动消费topic名称为maxwell的kafka consumer：


kafka-console-consumer.sh --zookeeper slave2:2181 --topic maxwell --from-beginning


7 在slave2上启动maxwell进程，将解析后的json数据输出到kafka：


bin/maxwell --user='maxwell' --password='111111' --host='192.168.175.21' --producer=kafka --kafka.bootstrap.servers=192.168.175.21:9092 --kafka_topic=maxwell


注意：启动之前，需要将输出到stdout上的maxwell进程停掉，否则会报错。


同样，启动成功后，会输出Binlog连接成功的日志信息。如下：


16:22:19,127 INFO  AppInfoParser - Kafka version : 1.0.0
16:22:19,129 INFO  AppInfoParser - Kafka commitId : aaa7af6d4a11b29d
16:22:19,391 INFO  Maxwell - Maxwell v1.21.1 is booting (MaxwellKafkaProducer), starting at Position[BinlogPosition[master.000002:1084309], lastHeartbeat=1555143612778]
16:22:20,916 INFO  MysqlSavedSchema - Restoring schema id 2 (last modified at Position[BinlogPosition[master.000001:40754], lastHeartbeat=1555112822550])
16:22:22,861 INFO  MysqlSavedSchema - Restoring schema id 1 (last modified at Position[BinlogPosition[master.000001:3816], lastHeartbeat=0])
16:22:23,178 INFO  MysqlSavedSchema - beginning to play deltas...
16:22:23,192 INFO  MysqlSavedSchema - played 1 deltas in 7ms
16:22:23,482 INFO  BinlogConnectorReplicator - Setting initial binlog pos to: master.000002:1084309
16:22:23,887 INFO  BinaryLogClient - Connected to 192.168.175.21:3306 at master.000002/1084309 (sid:6379, cid:64)
16:22:23,894 INFO  BinlogConnectorLifecycleListener - Binlog connected.


8 验证


（1）在slave1中的root账号登录的mysql窗口中，执行一条更新操作：


MySQL [mysql]> update tb_dept set name='xiaofei123' where id=14;
Query OK, 1 row affected (0.02 sec)
Rows matched: 1  Changed: 1  Warnings: 0


（2）随机在消费topic=maxwell的kafka consumer中输出日志如下：


{"database":"mysql","table":"tb_dept","type":"update","ts":1555143780,"xid":6888,"commit":true,"data":{"Id":14,"Name":"xiaofei123","description":"sales"},"old":{"Name":"xiaofei"}}


到此，流程梳理完毕！


六 遗留问题：

1 mysql数据库


设置远程访问之后，在本地一直访问不了，尝试修改密码同样访问不了。猜测是由于设置远程访问时的%百分号影响的。故而这里访问数据库用的都是远程访问的，其实正常生产环境，我们也都是远程进行访问的。


ERROR 1045 (28000): Access denied for user 'root'@'localhost' (using password: YES)


2 kafka：


在将maxwell进程和kafka server运行在同一台虚拟机上时，启动kafka consumer时，经常报ConsumerRebalanceFailedException的异常，如下：没有找到具体原因，最终通过kafka server和maxwell不运行在一起将问题解决。其实生产环境，一般kafka都是有专用的集群，也都不会和maxwell运行在一起。


[2019-04-13 13:12:03,599] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$)
kafka.common.ConsumerRebalanceFailedException: console-consumer-72779_slave2-1555132312263-9d33f2d8 can't rebalance after 4 retries
    at kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:660)
    at kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$reinitializeConsumer(ZookeeperConsumerConnector.scala:967)
    at kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.<init>(ZookeeperConsumerConnector.scala:1001)
    at kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:163)
    at kafka.consumer.OldConsumer.<init>(BaseConsumer.scala:75)
    at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:63)
    at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:47)
    at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)


文章标签：
云消息队列 Kafka 版
云解析DNS
微服务引擎
云数据库 RDS MySQL 版
日志服务
消息中间件
分布式计算
关系型数据库
Linux
MySQL
Kafka
Spark
数据格式
数据库
JSON
关键词：
日志服务数据
日志服务binlog
mysql日志服务
云数据库 RDS MySQL 版日志
云数据库 RDS MySQL 版binlog
游客tmj4hgt7bfvas
目录
相关文章
蓝易云
|
8月前
|
SQL 运维 关系型数据库
深入探讨MySQL的二进制日志（binlog）选项
总结而言，对MySQL binlogs深度理解并妥善配置对数据库运维管理至关重要；它不仅关系到系统性能优化也是实现高可靠性架构设计必须考虑因素之一。通过精心规划与周密部署可以使得该机能充分发挥作用而避免潜在风险带来影响。
蓝易云
261 6
杨枝甘露z
|
10月前
高性能网络库设计之日志组件
高性能网络库设计之日志组件
杨枝甘露z
267 2
蓝易云
|
SQL 监控 关系型数据库
MySQL日志分析：binlog、redolog、undolog三大日志的深度探讨。
数据库管理其实和写小说一样，需要规划，需要修订，也需要有能力回滚。理解这些日志的作用与优化，就像把握写作工具的使用与运用，为我们的数据库保驾护航。
蓝易云
824 23
游客bmwhmm6u34qx6
|
SQL 运维 关系型数据库
MySQL Binlog 日志查看方法及查看内容解析
本文介绍了 MySQL 的 Binlog（二进制日志）功能及其使用方法。Binlog 记录了数据库的所有数据变更操作，如 INSERT、UPDATE 和 DELETE，对数据恢复、主从复制和审计至关重要。文章详细说明了如何开启 Binlog 功能、查看当前日志文件及内容，并解析了常见的事件类型，包括 Format_desc、Query、Table_map、Write_rows、Update_rows 和 Delete_rows 等，帮助用户掌握数据库变化历史，提升维护和排障能力。
游客bmwhmm6u34qx6
5954 6
运营研究坊
|
SQL 存储 关系型数据库
简单聊聊MySQL的三大日志（Redo Log、Binlog和Undo Log）各有什么区别
在MySQL数据库管理中，理解Redo Log（重做日志）、Binlog（二进制日志）和Undo Log（回滚日志）至关重要。Redo Log确保数据持久性和崩溃恢复；Binlog用于主从复制和数据恢复，记录逻辑操作；Undo Log支持事务的原子性和隔离性，实现回滚与MVCC。三者协同工作，保障事务ACID特性。文章还详细解析了日志写入流程及可能的异常情况，帮助深入理解数据库日志机制。
运营研究坊
1855 0
成喆
日志服务数据加工最佳实践: 多子键为数组的复杂JSON加工
程序构建的日志经常会以一种统计性质的JSON格式写入, 通常其包含一个基础信息, 以及多个子健为数组的形式. 本篇如何使用日志服务数据加工处理多子键为数组的复杂JSON.
成喆
1278 0
蓝染-惣右介
|
XML 安全 Java
【日志框架整合】Slf4j、Log4j、Log4j2、Logback配置模板
本文介绍了Java日志框架的基本概念和使用方法，重点讨论了SLF4J、Log4j、Logback和Log4j2之间的关系及其性能对比。SLF4J作为一个日志抽象层，允许开发者使用统一的日志接口，而Log4j、Logback和Log4j2则是具体的日志实现框架。Log4j2在性能上优于Logback，推荐在新项目中使用。文章还详细说明了如何在Spring Boot项目中配置Log4j2和Logback，以及如何使用Lombok简化日志记录。最后，提供了一些日志配置的最佳实践，包括滚动日志、统一日志格式和提高日志性能的方法。
蓝染-惣右介
4897 32
【日志框架整合】Slf4j、Log4j、Log4j2、Logback配置模板
运维有小邓
|
监控 安全 Apache
什么是Apache日志？为什么Apache日志分析很重要？
Apache是全球广泛使用的Web服务器软件，支持超过30%的活跃网站。它通过接收和处理HTTP请求，与后端服务器通信，返回响应并记录日志，确保网页请求的快速准确处理。Apache日志分为访问日志和错误日志，对提升用户体验、保障安全及优化性能至关重要。EventLog Analyzer等工具可有效管理和分析这些日志，增强Web服务的安全性和可靠性。
运维有小邓
596 9
阿里云云原生
|
监控 容灾 算法
阿里云 SLS 多云日志接入最佳实践：链路、成本与高可用性优化
本文探讨了如何高效、经济且可靠地将海外应用与基础设施日志统一采集至阿里云日志服务(SLS)，解决全球化业务扩展中的关键挑战。重点介绍了高性能日志采集Agent（iLogtail/LoongCollector）在海外场景的应用，推荐使用LoongCollector以获得更优的稳定性和网络容错能力。同时分析了多种网络接入方案，包括公网直连、全球加速优化、阿里云内网及专线/CEN/VPN接入等，并提供了成本优化策略和多目标发送配置指导，帮助企业构建稳定、低成本、高可用的全球日志系统。
阿里云云原生
1175 55

热门文章

最新文章

  • 1
    阿里云DTS踩坑经验分享系列｜如何使用DTS进行MySQL->ClickHouse同步
  • 2
    四种mysql可视化操作工具的对比
  • 3
    Dataphin实现MaxCompute外表数据快速批量同步至ADB MySQL
  • 4
    Linux安装部署MySQL5.7（企业常用版）超详细1
  • 5
    MySQL中的引号和反引号（``和 ‘‘）
  • 6
    磁盘高负荷把MYSQL拖垮（请用火狐浏览器浏览）
  • 7
    【MySQL】计算 TPS,QPS 的python 脚本
  • 8
    MySQL索引最左匹配原则及优化原理
  • 9
    mysql if对数据进行处理 having对数据进行查询 thinkphp中的exp支持更复杂的where查询
  • 10
    ubuntu下apache和mysql的命令
  • 1
    【赵渝强老师】大数据日志采集引擎Flume
    434
  • 2
    16 倍性能提升，成本降低 98%！ 解读 SLS 向量索引架构升级改造
    612
  • 3
    防止员工泄密软件中文件访问日志管理的 Go 语言 B + 树算法
    228
  • 4
    局域网网络监控软件的设备连接日志哈希表 C++ 语言算法
    600
  • 5
    Python日志模块配置：从print到logging的优雅升级指南
    637
  • 6
    SQL日志优化策略：提升数据库日志记录效率
    396
  • 7
    程序出错瞎找？教你写“会说话”的错误日志，秒定位原因
    582
  • 8
    SLS Copilot 实践：基于 SLS 灵活构建 LLM 应用的数据基础设施
    2665
  • 9
    EventLog Analyzer：高效的Web服务器日志监控与审计解决方案
    424
  • 10
    5 个 IIS 日志记录最佳实践
    443

    • 相关课程

    更多
  • 云数据库MySQL版快速上手教程
  • 阿里云云原生数据仓库AnalyticDB MySQL版 使用教程
  • MySQL实战进阶
  • 阿里云数据库产品家族及特性
  • 数据库及SQL/MySQL基础
  • 云数据库MySQL快速入门

    • 相关电子书

    更多
  • Kubernetes下日志实时采集、存储与计算实践
  • 日志数据采集与分析对接
  • 智能化日志中心

    • 推荐镜像

    更多
  • mysql
    • 下一篇
    阿里云正式发布 Agentic 代码安全：AI驱动的双Agent协同引擎