# 基于binlog的MySQL数据同步到MRS集群中 本章节为您介绍使用Maxwell同步工具将线下基于binlog的数据迁移到MRS Kafka集群中的指导。 Maxwell是一个开源程序（），通过读取MySQL的binlog日志，将增删改等操作转为JSON格式发送到输出端（如控制台/文件/Kafka等）。Maxwell可部署在MySQL机器上，也可独立部署在其他与MySQL网络可通的机器上。 Maxwell运行在Linux服务器上，常见支持的操作系统有EulerOS、Ubuntu、Debian、CentOS、OpenSUSE等，且需要Java 1.8+支持。 同步数据具体流程如下。 1. [配置MySQL] 2. [安装Maxwell] 3. [配置Maxwell] 4. [启动Maxwell] 5. [验证Maxwell] 6. [停止Maxwell] 7. [Maxwell生成的数据格式及常见字段含义] #### 配置MySQL 1. 开启binlog，在MySQL中打开my.cnf文件，在\[mysqld\] 区块检查是否配置server_id，log-bin与binlog_format，如果没有配置请执行如下命令添加配置项并重启MySQL，如果已经配置则忽略此步骤。 ``` $ vi my.cnf [mysqld] server_id=1 log-bin=master binlog_format=row ``` **如果使用了云数据库 RDS for MySQL，建议配置Binlog的保留时长，防止本地Binlog数据被删除，具体操作可参考[设置RDS for MySQL本地Binlog日志清理](https://support.huaweicloud.com/usermanual-rds-mysql/rds_05_0037.html)。** 2. Maxwell需要连接MySQL，并创建一个名称为maxwell的数据库存储元数据，且需要能访问需要同步的数据库，所以建议新创建一个MySQL用户专门用来给Maxwell使用。使用root登录MySQL之后，执行如下命令创建maxwell用户（其中XXXXXX是密码，请修改为实际值）。命令中如果携带认证密码信息可能存在安全风险，在执行命令前建议关闭系统的history命令记录功能，避免信息泄露。 - 如果Maxwell程序部署在非MySQL机器上，则创建的maxwell用户需要有远程登录数据库的权限，此时创建命令为： ``` mysql> GRANT ALL on maxwell.* to 'maxwell'@'%' identified by 'XXXXXX'; ``` ``` mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%'; ``` - 如果Maxwell部署在MySQL机器上，则创建的maxwell用户可以设置为只能在本机登录数据库，此时创建命令为： ``` mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'localhost' identified by 'XXXXXX'; ``` ``` mysql> GRANT ALL on maxwell.* to 'maxwell'@'localhost'; ``` #### 安装Maxwell 1. 下载安装包，下载路径为，选择名为maxwell-XXX.tar.gz的二进制文件下载，其中XXX为版本号。 2. 将tar.gz包上传到任意目录下（本示例路径为Master节点的"/opt"）。 3. 登录部署Maxwell的服务器，并执行如下命令进入tar.gz包所在目录。 ``` cd /opt ``` 4. 执行如下命令解压"maxwell-XXX.tar.gz"压缩包，并进入"maxwell-XXX"文件夹。 ``` tar -zxvf maxwell-XXX.tar.gz ``` ``` cd maxwell-XXX ``` #### 配置Maxwell 在maxwell-XXX文件夹下如果有conf目录则配置config.properties文件，配置项说明请参见[表1]。如果没有conf目录，则在maxwell-XXX文件夹下将config.properties.example修改成config.properties。 表1Maxwell配置项说明 | 配置项 | 是否必填 | 说明 | 默认值 | |:---|:---|:---|:---| | user | 是 | 连接MySQL的用户名，即[2]中新创建的用户。 | - | | password | 是 | 连接MySQL的密码。配置文件中包含认证密码信息可能存在安全风险，建议当前场景执行完毕后删除相关配置文件或加强安全管理。 | - | | host | 否 | MySQL地址。 | localhost | | port | 否 | MySQL端口。 | 3306 | | log_level | 否 | 日志打印级别，可选值为： - debug - info - warn - error | info | | output_ddl | 否 | 是否发送DDL（数据库与数据表的定义修改）事件： - true：发送DDL事件 - false：不发送DDL事件 | false | | producer | 是 | 生产者类型，配置为"kafka"。 - stdout：将生成的事件打印在日志中 - kafka：将生成的事件发送到Kafka | stdout | | producer_partition_by | 否 | 分区策略，用来确保相同一类的数据写入到Kafka同一分区。 - database：使用数据库名称做分区，保证同一个数据库的事件写入到Kafka同一个分区中。 - table：使用表名称做分区，保证同一个表的事件写入到Kafka同一个分区中。 | database | | ignore_producer_error | 否 | 是否忽略生产者发送数据失败的错误。 - true：在日志中打印错误信息并跳过错误的数据，程序继续运行。 - false：在日志中打印错误信息并终止程序。 | true | | metrics_slf4j_interval | 否 | 在日志中输出上传kafka成功与失败数据的数量统计的时间间隔，单位为秒。 | 60 | | kafka.bootstrap.servers | 是 | Kafka代理节点地址，配置形式为HOST:PORT\[,HOST:PORT\] | - | | kafka_topic | 否 | 写入Kafka的Topic名称。 | maxwell | | dead_letter_topic | 否 | 当发送某条记录出错时，记录该条出错记录主键的Kafka Topic。 | - | | kafka_version | 否 | Maxwell使用的Kafka producer版本号，不能在config.properties中配置，需要在启动命令时用-- kafka_version xxx参数传入。 | - | | kafka_partition_hash | 否 | 划分Kafka Topic partition的算法，支持default或murmur3。 | default | | kafka_key_format | 否 | Kafka record的key生成方式，支持array或Hash。 | Hash | | ddl_kafka_topic | 否 | 当output_ddl配置为true时，DDL操作写入的Topic。 | {kafka_topic} | | filter | 否 | 过滤数据库或表。 - 如果只想采集mydatabase的库，可以配置为： ``` exclude: *.*,include: mydatabase.* ``` - 如果只想采集mydatabase.mytable的表，可以配置为： ``` exclude: *.*,include: mydatabase.mytable ``` - 如果只想采集mydatabase库下的mytable，mydate_123, mydate_456表，可以配置为： ``` exclude: *.*,include: mydatabase.mytable, include: mydatabase./mydate_\\d*/ ``` | - | #### 启动Maxwell 1. 登录Maxwell所在的服务器。 2. 执行如下命令进入Maxwell安装目录。 ``` cd /opt/maxwell-1.21.0/ ``` ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/note_3.0-zh-cn.png) 如果是初次使用Maxwell，建议将conf/config.properties中的log_level改为debug(调试级别)，以便观察启动之后是否能正常从MySQL获取数据并发送到kafka，当整个流程调试通过之后，再把log_level修改为info，然后先停止再启动Maxwell生效。 ``` # log level [debug | info | warn | error] log_level=debug ``` 3. 执行如下命令启动Maxwell。 ``` source 客户端安装路径/bigdata_env ``` ``` bin/Maxwell ``` ``` bin/maxwell --user='maxwell' --password='XXXXXX' --host='127.0.0.1' \ ``` ``` --producer=kafka --kafka.bootstrap.servers=kafkahost:9092 --kafka_topic=Maxwell ``` 其中，user，password和host分别表示MySQL的用户名，密码和IP地址，这三个参数可以通过修改配置项配置也可以通过上述命令配置，kafkahost为流式集群的Core节点的IP地址。 显示类似如下信息，表示Maxwell启动成功。 ``` Success to start Maxwell [78092]. ``` #### 验证Maxwell 1. 登录Maxwell所在的服务器。 2. 查看日志。如果日志里面没有ERROR日志，且有打印如下日志，表示与MySQL连接正常。 ``` BinlogConnectorLifecycleListener - Binlog connected. ``` 3. 登录MySQL数据库，对测试数据进行更新/创建/删除等操作。操作语句可以参考如下示例。 ``` -- 创建库 create database test; -- 创建表 create table test.e ( id int(10) not null primary key auto_increment, m double, c timestamp(6), comment varchar(255) charset 'latin1' ); -- 增加记录 insert into test.e set m = 4.2341, c = now(3), comment = 'I am a creature of light.'; -- 更新记录 update test.e set m = 5.444, c = now(3) where id = 1; -- 删除记录 delete from test.e where id = 1; -- 修改表 alter table test.e add column torvalds bigint unsigned after m; -- 删除表 drop table test.e; -- 删除库 drop database test; ``` 4. 观察Maxwell的日志输出，如果没有WARN/ERROR打印，则表示Maxwell安装配置正常。 如果要确定数据是否成功上传，可设置config.properties中的log_level为debug，则数据上传成功时会立刻打印如下JSON格式数据，具体字段含义请参考[Maxwell生成的数据格式及常见字段含义]。 ``` {"database":"test","table":"e","type":"insert","ts":1541150929,"xid":60556,"commit":true,"data":{"id":1,"m":4.2341,"c":"2018-11-02 09:28:49.297000","comment":"I am a creature of light."}} …… ``` ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/note_3.0-zh-cn.png) 当整个流程调试通过之后，可以把config.properties文件中的配置项log_level修改为info，减少日志打印量，并重启Maxwell。 ``` # log level [debug | info | warn | error] log_level=info ``` #### 停止Maxwell 1. 登录Maxwell所在的服务器。 2. 执行如下命令，获取Maxwell的进程标识（PID）。输出的第二个字段即为PID。 ``` ps -ef | grep Maxwell | grep -v grep ``` 3. 执行如下命令，强制停止Maxwell进程。 ``` kill -9 PID ``` #### Maxwell生成的数据格式及常见字段含义 Maxwell生成的数据格式为JSON，常见字段含义如下： - type：操作类型，包含database-create，database-drop，table-create，table-drop，table-alter，insert，update，delete。 - database：操作的数据库名称。 - ts：操作时间，13位时间戳。 - table：操作的表名。 - data：数据增加/删除/修改之后的内容。 - old：数据修改前的内容或者表修改前的结构定义。 - sql：DDL操作的SQL语句。 - def：表创建与表修改的结构定义。 - xid：事务唯一ID。 - commit：数据增加/删除/修改操作是否已提交。