专栏首页小旋锋的大数据专栏MySQL Binlog 解析工具 Maxwell 详解

MySQL Binlog 解析工具 Maxwell 详解

maxwell 简介

Maxwell是一个能实时读取MySQL二进制日志binlog,并生成 JSON 格式的消息,作为生产者发送给 Kafka,Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件或其它平台的应用程序。它的常见应用场景有ETL、维护缓存、收集表级别的dml指标、增量到搜索引擎、数据分区迁移、切库binlog回滚方案等。官网(http://maxwells-daemon.io)、GitHub(https://github.com/zendesk/maxwell)

Maxwell主要提供了下列功能:

  • 支持 SELECT*FROM table 的方式进行全量数据初始化
  • 支持在主库发生failover后,自动恢复binlog位置(GTID)
  • 可以对数据进行分区,解决数据倾斜问题,发送到kafka的数据支持database、table、column等级别的数据分区
  • 工作方式是伪装为Slave,接收binlog events,然后根据schemas信息拼装,可以接受ddl、xid、row等各种event

除了Maxwell外,目前常用的MySQL Binlog解析工具主要有阿里的canal、mysql_streamer,三个工具对比如下:

Canal 由Java开发,分为服务端和客户端,拥有众多的衍生应用,性能稳定,功能强大;canal 需要自己编写客户端来消费canal解析到的数据。

maxwell相对于canal的优势是使用简单,它直接将数据变更输出为json字符串,不需要再编写客户端。

快速开始

首先MySQL需要先启用binlog,关于什么是MySQL binlog,可以参考文章《MySQL Binlog 介绍

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

创建Maxwell用户,并赋予 maxwell 库的一些权限

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

使用 maxwell 之前需要先启动 kafka

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.1.0/kafka_2.11-2.1.0.tgztar -xzf kafka_2.11-2.1.0.tgzcd kafka_2.11-2.1.0# 启动Zookeeperbin/zookeeper-server-start.sh config/zookeeper.properties

单机启动 kafka 之前,需要修改一下配置文件,打开配置文件 vi config/server.properties,在文件最后加入 advertised.host.name 的配置,值为 kafka 所在机器的IP

advertised.host.name=10.100.97.246

不然后面通过 docker 启动 maxwell 将会报异常(其中的 hadoop2 是我的主机名)

17:45:21,446 DEBUG NetworkClient - [Producer clientId=producer-1] Error connecting to node hadoop2:9092 (id: 0 rack: null)java.io.IOException: Can't resolve address: hadoop2:9092        at org.apache.kafka.common.network.Selector.connect(Selector.java:217) ~[kafka-clients-1.0.0.jar:?]        at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:793) [kafka-clients-1.0.0.jar:?]        at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:230) [kafka-clients-1.0.0.jar:?]        at org.apache.kafka.clients.producer.internals.Sender.sendProducerData(Sender.java:263) [kafka-clients-1.0.0.jar:?]        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:238) [kafka-clients-1.0.0.jar:?]        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:176) [kafka-clients-1.0.0.jar:?]        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_181]Caused by: java.nio.channels.UnresolvedAddressException        at sun.nio.ch.Net.checkAddress(Net.java:101) ~[?:1.8.0_181]        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622) ~[?:1.8.0_181]        at org.apache.kafka.common.network.Selector.connect(Selector.java:214) ~[kafka-clients-1.0.0.jar:?]        ... 6 more

接着可以启动 kafka

bin/kafka-server-start.sh config/server.properties

测试 kafka

# 创建一个 topicbin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
# 列出所有 topicbin/kafka-topics.sh --list --zookeeper localhost:2181
# 启动一个生产者,然后随意发送一些消息bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testThis is a messageThis is another message
# 在另一个终端启动一下消费者,观察所消费的消息bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginningThis is a messageThis is another message

通过 docker 快速安装并使用 Maxwell (当然之前需要自行安装 docker)

# 拉取镜像 docker pull zendesk/maxwell
# 启动maxwell,并将解析出的binlog输出到控制台docker run -ti --rm zendesk/maxwell bin/maxwell --user='maxwell' --password='123456' --host='10.100.97.246' --producer=stdout

测试Maxwell,首先创建一张简单的表,然后增改删数据

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;
insert into test values(1,22,"小旋锋");update test set name='whirly' where id=1;delete from test where id=1;

观察docker控制台的输出,从输出的日志中可以看出Maxwell解析出的binlog的JSON字符串的格式

{"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"}}

输出到 Kafka,关闭 docker,重新设置启动参数

docker run -it --rm zendesk/maxwell bin/maxwell --user='maxwell' \    --password='123456' --host='10.100.97.246' --producer=kafka \    --kafka.bootstrap.servers='10.100.97.246:9092' --kafka_topic=maxwell --log_level=debug

然后启动一个消费者来消费 maxwell topic的消息,观察其输出;再一次执行增改删数据的SQL,仍然可以得到相同的输出

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic maxwell

输出JSON字符串的格式

  • data 最新的数据,修改后的数据
  • old 旧数据,修改前的数据
  • type 操作类型,有insert, update, delete, database-create, database-alter, database-drop, table-create, table-alter, table-drop,bootstrap-insert,int(未知类型)
  • xid 事务id
  • commit 同一个xid代表同一个事务,事务的最后一条语句会有commit,可以利用这个重现事务
  • server_id
  • thread_id
  • 运行程序时添加参数--output_ddl,可以捕捉到ddl语句
  • datetime列会输出为"YYYY-MM-DD hh:mm:ss",如果遇到"0000-00-00 00:00:00"会原样输出
  • maxwell支持多种编码,但仅输出utf8编码
  • maxwell的TIMESTAMP总是作为UTC处理,如果要调整为自己的时区,需要在后端逻辑上进行处理

与输出格式相关的配置如下

选项

参数值

描述

默认值

output_binlog_position

BOOLEAN

是否包含 binlog position

false

output_gtid_position

BOOLEAN

是否包含 gtid position

false

output_commit_info

BOOLEAN

是否包含 commit and xid

true

output_xoffset

BOOLEAN

是否包含 virtual tx-row offset

false

output_nulls

BOOLEAN

是否包含值为NULL的字段

true

output_server_id

BOOLEAN

是否包含 server_id

false

output_thread_id

BOOLEAN

是否包含 thread_id

false

output_schema_id

BOOLEAN

是否包含 schema_id

false

output_row_query

BOOLEAN

是否包含 INSERT/UPDATE/DELETE 语句. Mysql需要开启 binlog_rows_query_log_events

false

output_ddl

BOOLEAN

是否包含 DDL (table-alter, table-create, etc) events

false

output_null_zerodates

BOOLEAN

是否将 '0000-00-00' 转换为 null?

false

进阶使用

基本的配置

选项

参数值

描述

默认值

config

配置文件 config.properties 的路径

log_level

`[debug

info

warn

daemon

指定Maxwell实例作为守护进程到后台运行

env_config_prefix

STRING

匹配该前缀的环境变量将被视为配置值

可以把Maxwell的启动参数写到一个配置文件 config.properties 中,然后通过 config 选项指定, bin/maxwell--config config.properties

user=maxwellpassword=123456host=10.100.97.246producer=kafkakafka.bootstrap.servers=10.100.97.246:9092kafka_topic=maxwell

mysql 配置选项

Maxwell 根据用途将 MySQL 划分为3种角色:

host:主机,建maxwell库表,存储捕获到的schema等信息

  • 主要有六张表,bootstrap用于数据初始化,schemas记录所有的binlog文件信息,databases记录了所有的数据库信息,tables记录了所有的表信息,columns记录了所有的字段信息,positions记录了读取binlog的位移信息,heartbeats记录了心跳信息

replication_host:复制主机,Event监听,读取该主机binlog

  • hostreplication_host分开,可以避免 replication_user 往生产库里写数据

schema_host:schema主机,捕获表结构schema的主机

  • binlog里面没有字段信息,所以maxwell需要从数据库查出schema,存起来。
  • schema_host一般用不到,但在binlog-proxy场景下就很实用。比如要将已经离线的binlog通过maxwell生成json流,于是自建一个mysql server里面没有结构,只用于发送binlog,此时表机构就可以制动从 schema_host 获取。

通常,这三个主机都是同一个, schema_host 只在有 replication_host 的时候使用。

与MySQL相关的有下列配置

选项

参数值

描述

默认值

host

STRING

mysql 地址

localhost

user

STRING

mysql 用户名

password

STRING

mysql 密码

(no password)

port

INT

mysql 端口 3306

jdbc_options

STRING

mysql jdbc connection options

DEFAULT_JDBC_OPTS

ssl

SSL_OPT

SSL behavior for mysql cx

DISABLED

schema_database

STRING

Maxwell用于维护的schema和position将使用的数据库

maxwell

client_id

STRING

用于标识Maxwell实例的唯一字符串

maxwell

replica_server_id

LONG

用于标识Maxwell实例的唯一数字

6379 (see notes)

master_recovery

BOOLEAN

enable experimental master recovery code

false

gtid_mode

BOOLEAN

是否开启基于GTID的复制

false

recapture_schema

BOOLEAN

重新捕获最新的表结构(schema),不可在 config.properties中配置

false

replication_host

STRING

server to replicate from. See split server roles

schema-store host

replication_password

STRING

password on replication server

(none)

replication_port

INT

port on replication server

3306

replication_user

STRING

user on replication server

replication_ssl

SSL_OPT

SSL behavior for replication cx cx

DISABLED

schema_host

STRING

server to capture schema from. See split server roles

schema-store host

schema_password

STRING

password on schema-capture server

(none)

schema_port

INT

port on schema-capture server

3306

schema_user

STRING

user on schema-capture server

schema_ssl

SSL_OPT

SSL behavior for schema-capture server

DISABLED

生产者的配置

仅介绍kafka,其他的生产者的配置详见官方文档。

kafka是maxwell支持最完善的一个生产者,并且内置了多个版本的kafka客户端(0.8.2.2, 0.9.0.1, 0.10.0.1, 0.10.2.1 or 0.11.0.1, 1.0.0.),默认 kafka_version=1.0.0(当前Maxwell版本1.20.0)

Maxwell 会将消息投递到Kafka的Topic中,该Topic由 kafka_topic 选项指定,默认值为 maxwell,除了指定为静态的Topic,还可以指定为动态的,譬如 namespace_%{database}_%{table}%{database}%{table} 将被具体的消息的 database 和 table 替换。

Maxwell 读取配置时,如果配置项是以 kafka. 开头,那么该配置将设置到 Kafka Producer 客户端的连接参数中去,譬如

kafka.acks = 1kafka.compression.type = snappykafka.retries=5

下面是Maxwell通用生产者和Kafka生产者的配置参数

选项

参数值

描述

默认值

producer

PRODUCER_TYPE

生产者类型

stdout

custom_producer.factory

CLASS_NAME

自定义消费者的工厂类

producer_ack_timeout

PRODUCER_ACK_TIMEOUT

异步消费认为消息丢失的超时时间(毫秒ms)

producer_partition_by

PARTITION_BY

输入到kafka/kinesis的分区函数

database

producer_partition_columns

STRING

若按列分区,以逗号分隔的列名称

producer_partition_by_fallback

PARTITION_BY_FALLBACK

producer_partition_by=column时需要,当列不存在是使用

ignore_producer_error

BOOLEAN

为false时,在kafka/kinesis发生错误时退出程序;为true时,仅记录日志 See also dead_letter_topic

true

kafka.bootstrap.servers

STRING

kafka 集群列表, HOST:PORT[,HOST:PORT]

kafka_topic

STRING

kafka topic

maxwell

dead_letter_topic

STRING

详见官方文档

kafka_version

KAFKA_VERSION

指定maxwell的 kafka 生产者客户端版本,不可在config.properties中配置

0.11.0.1

kafka_partition_hash

`[default

murmur3]`

选择kafka分区时使用的hash方法

kafka_key_format

`[array

hash]`

how maxwell outputs kafka keys, either a hash or an array of hashes

ddl_kafka_topic

STRING

当 output_ddl为true时, 所有DDL的消息都将投递到该topic

kafka_topic

过滤器配置

Maxwell 可以通过 --filter 配置项来指定过滤规则,通过 exclude 排除,通过 include 包含,值可以为具体的数据库、数据表、数据列,甚至用 Javascript 来定义复杂的过滤规则;可以用正则表达式描述,有几个来自官网的例子

# 仅匹配foodb数据库的tbl表和所有table_数字的表--filter='exclude: foodb.*, include: foodb.tbl, include: foodb./table_\d+/'# 排除所有库所有表,仅匹配db1数据库--filter = 'exclude: *.*, include: db1.*'# 排除含db.tbl.col列值为reject的所有更新--filter = 'exclude: db.tbl.col = reject'# 排除任何包含col_a列的更新--filter = 'exclude: *.*.col_a = *'# blacklist 黑名单,完全排除bad_db数据库,若要恢复,必须删除maxwell库--filter = 'blacklist: bad_db.*' 

数据初始化

Maxwell 启动后将从maxwell库中获取上一次停止时position,从该断点处开始读取binlog。如果binlog已经清除了,那么怎样可以通过maxwell把整张表都复制出来呢?也就是数据初始化该怎么做?

对整张表进行操作,人为地产生binlog?譬如找一个不影响业务的字段譬如update_time,然后加一秒,再减一秒?

update test set update_time = DATE_ADD(update_time,intever 1 second);update test set update_time = DATE_ADD(update_time,intever -1 second);

这样明显存在几个大问题:

  • 不存在一个不重要的字段怎么办?每个字段都很重要,不能随便地修改!
  • 如果整张表很大,修改的过程耗时很长,影响了业务!
  • 将产生大量非业务的binlog!

针对数据初始化的问题,Maxwell 提供了一个命令工具 maxwell-bootstrap 帮助我们完成数据初始化, maxwell-bootstrap 是基于 SELECT*FROM table 的方式进行全量数据初始化,不会产生多余的binlog!

这个工具有下面这些参数:

日志级别(DEBUG, INFO, WARN or ERROR)

实验一番,下面将引导 test 数据库中 test 表,首先是准备几条测试用的数据

INSERT INTO `test` VALUES (1, 1, '1');INSERT INTO `test` VALUES (2, 2, '2');INSERT INTO `test` VALUES (3, 3, '3');INSERT INTO `test` VALUES (4, 4, '4');

然后 reset master; 清空binlog,删除 maxwell 库中的表。接着使用快速开始中的命令,启动Kafka、Maxwell和Kafka消费者,然后启动 maxwell-bootstrap

docker run -it --rm zendesk/maxwell bin/maxwell-bootstrap --user maxwell  \    --password 123456 --host 10.100.97.246  --database test --table test --client_id maxwell

注意: --bootstrapper=sync 时,在处理bootstrap时,会阻塞正常的binlog解析; --bootstrapper=async 时,不会阻塞。

也可以执行下面的SQL,在 maxwell.bootstrap 表中插入记录,手动触发

insert into maxwell.bootstrap (database_name, table_name) values ('test', 'test');

就可以在 kafka 消费者端看见引导过来的数据了

{"database":"maxwell","table":"bootstrap","type":"insert","ts":1552199115,"xid":36738,"commit":true,"data":{"id":3,"database_name":"test","table_name":"test","where_clause":null,"is_complete":0,"inserted_rows":0,"total_rows":0,"created_at":null,"started_at":null,"completed_at":null,"binlog_file":null,"binlog_position":0,"client_id":"maxwell"}}{"database":"test","table":"test","type":"bootstrap-start","ts":1552199115,"data":{}}{"database":"test","table":"test","type":"bootstrap-insert","ts":1552199115,"data":{"id":1,"age":1,"name":"1"}}{"database":"test","table":"test","type":"bootstrap-insert","ts":1552199115,"data":{"id":2,"age":2,"name":"2"}}{"database":"test","table":"test","type":"bootstrap-insert","ts":1552199115,"data":{"id":3,"age":3,"name":"3"}}{"database":"test","table":"test","type":"bootstrap-insert","ts":1552199115,"data":{"id":4,"age":4,"name":"4"}}{"database":"maxwell","table":"bootstrap","type":"update","ts":1552199115,"xid":36756,"commit":true,"data":{"id":3,"database_name":"test","table_name":"test","where_clause":null,"is_complete":1,"inserted_rows":4,"total_rows":0,"created_at":null,"started_at":"2019-03-10 14:25:15","completed_at":"2019-03-10 14:25:15","binlog_file":"mysql-bin.000001","binlog_position":64446,"client_id":"maxwell"},"old":{"is_complete":0,"inserted_rows":1,"completed_at":null}}{"database":"test","table":"test","type":"bootstrap-complete","ts":1552199115,"data":{}}

中间的4条便是 test.test 的binlog数据了,注意这里的 type 是 bootstrap-insert,而不是 insert

然后再一次查看binlog, show binlog events;,会发现只有与 maxwell 相关的binlog,并没有 test.test 相关的binlog,所以 maxwell-bootstrap 命令并不会产生多余的 binlog,当数据表的数量很大时,这个好处会更加明显

Bootstrap 的过程是 bootstrap-start->bootstrap-insert->bootstrap-complete,其中,start和complete的data字段为空,不携带数据。

在进行bootstrap过程中,如果maxwell崩溃,重启时,bootstrap会完全重新开始,不管之前进行到多少,若不希望这样,可以到数据库中设置 is_complete 字段值为1(表示完成),或者删除该行

Maxwell监控

Maxwell 提供了 baselogging mechanism,JMX,HTTPorbypush toDatadog 这四种监控方式,与监控相关的配置项有下列这些:

选项

参数值

描述

默认值

metrics_prefix

STRING

指标的前缀

MaxwellMetrics

metrics_type

`[slf4j

jmx

http

metrics_jvm

BOOLEAN

是否收集JVM信息

false

metrics_slf4j_interval

SECONDS

将指标记录到日志的频率, metrics_type须配置为slf4j

60

http_port

INT

metrics_type为http时,发布指标绑定的端口

8080

http_path_prefix

STRING

http的路径前缀

/

http_bind_address

STRING

http发布指标绑定的地址

all addresses

http_diagnostic

BOOLEAN

http是否开启diagnostic后缀

false

http_diagnostic_timeout

MILLISECONDS

http diagnostic 响应超时时间

10000

metrics_datadog_type

`[udp

http]`

metrics_type为datadog时发布指标的方式

metrics_datadog_tags

STRING

提供给 datadog 的标签,如 tag1:value1,tag2:value2

metrics_datadog_interval

INT

推指标到datadog的频率,单位秒

60

metrics_datadog_apikey

STRING

当 metrics_datadog_type=http 时datadog用的api key

metrics_datadog_host

STRING

当 metrics_datadog_type=udp时推指标的目标地址

localhost

metrics_datadog_port

INT

当 metrics_datadog_type=udp 时推指标的端口

8125

具体可以得到哪些监控指标呢?有如下,注意所有指标都预先配置了指标前缀 metrics_prefix

指标

类型

说明

messages.succeeded

Counters

成功发送到kafka的消息数量

messages.failed

Counters

发送失败的消息数量

row.count

Counters

已处理的binlog行数,注意并非所有binlog都发往kafka

messages.succeeded.meter

Meters

消息成功发送到Kafka的速率

messages.failed.meter

Meters

消息发送失败到kafka的速率

row.meter

Meters

行(row)从binlog连接器到达maxwell的速率

replication.lag

Gauges

从数据库事务提交到Maxwell处理该事务之间所用的时间(毫秒)

inflightmessages.count

Gauges

当前正在处理的消息数(等待来自目的地的确认,或在消息之前)

message.publish.time

Timers

向kafka发送record所用的时间(毫秒)

message.publish.age

Timers

从数据库产生事件到发送到Kafka之间的时间(毫秒),精确度为+/-500ms

replication.queue.time

Timers

将一个binlog事件送到处理队列所用的时间(毫秒)

上述有些指标为kafka特有的,并不支持所有的生产者。

实验一番,通过 http 方式获取监控指标

docker run -p 8080:8080 -it --rm zendesk/maxwell bin/maxwell --user='maxwell' \    --password='123456' --host='10.100.97.246' --producer=kafka \    --kafka.bootstrap.servers='10.100.97.246:9092' --kafka_topic=maxwell --log_level=debug \    --metrics_type=http  --metrics_jvm=true --http_port=8080

上面的配置大部分与前面的相同,不同的有 -p8080:8080 docker端口映射,以及 --metrics_type=http--metrics_jvm=true--http_port=8080,配置了通过http方式发布指标,启用收集JVM信息,端口为8080,之后可以通过 http://10.100.97.246:8080/metrics 便可获取所有的指标

http 方式有四种后缀,分别对应四种不同的格式

所有指标以JSON格式返回

如果是通过 JMX 的方式收集Maxwell监控指标,可以 JAVA_OPTS 环境变量配置JMX访问权限

export JAVA_OPTS="-Dcom.sun.management.jmxremote \-Dcom.sun.management.jmxremote.port=9010 \-Dcom.sun.management.jmxremote.local.only=false \-Dcom.sun.management.jmxremote.authenticate=false \-Dcom.sun.management.jmxremote.ssl=false \-Djava.rmi.server.hostname=10.100.97.246"

多个Maxwell实例

在不同的配置下,Maxwell可以在同一个主服务器上运行多个实例。如果希望让生产者以不同的配置运行,例如将来自不同组的表(table)的事件投递到不同的Topic中,这将非常有用。Maxwell的每个实例都必须配置一个唯一的client_id,以便区分的binlog位置。

GTID 支持

Maxwell 从1.8.0版本开始支持基于GTID的复制(GTID-based replication),在GTID模式下,Maxwell将在主机更改后透明地选择新的复制位置。

什么是GTID Replication?

GTID (Global Transaction ID) 是对于一个已提交事务的编号,并且是一个全局唯一的编号。

从 MySQL 5.6.5 开始新增了一种基于 GTID 的复制方式。通过 GTID 保证了每个在主库上提交的事务在集群中有一个唯一的ID。这种方式强化了数据库的主备一致性,故障恢复以及容错能力。

在原来基于二进制日志的复制中,从库需要告知主库要从哪个偏移量进行增量同步,如果指定错误会造成数据的遗漏,从而造成数据的不一致。借助GTID,在发生主备切换的情况下,MySQL的其它从库可以自动在新主库上找到正确的复制位置,这大大简化了复杂复制拓扑下集群的维护,也减少了人为设置复制位置发生误操作的风险。另外,基于GTID的复制可以忽略已经执行过的事务,减少了数据发生不一致的风险。

注意事项

timestamp column

maxwell对时间类型(datetime, timestamp, date)都是当做字符串处理的,这也是为了保证数据一致(比如 0000-00-0000:00:00这样的时间在timestamp里是非法的,但mysql却认,解析成java或者python类型就是null/None)。

如果MySQL表上的字段是 timestamp 类型,是有时区的概念,binlog解析出来的是标准UTC时间,但用户看到的是本地时间。比如 f_create_time timestamp 创建时间是北京时间 2018-01-0521:01:01,那么mysql实际存储的是 2018-01-0513:01:01,binlog里面也是这个时间字符串。如果不做消费者不做时区转换,会少8个小时。

与其每个客户端都要考虑这个问题,我觉得更合理的做法是提供时区参数,然后maxwell自动处理时区问题,否则要么客户端先需要知道哪些列是timestamp类型,或者连接上原库缓存上这些类型。

binary column

maxwell可以处理binary类型的列,如blob、varbinary,它的做法就是对二进制列使用 base64_encode,当做字符串输出到json。消费者拿到这个列数据后,不能直接拼装,需要 base64_decode

表结构不同步

如果是拿比较老的binlog,放到新的mysql server上去用maxwell拉去,有可能表结构已经发生了变化,比如binlog里面字段比 schema_host 里面的字段多一个。目前这种情况没有发现异常,比如阿里RDS默认会为 无主键无唯一索引的表,增加一个 __##alibaba_rds_rowid##__,在 show create tableschema 里面都看不到这个隐藏主键,但binlog里面会有,同步到从库。

另外我们有通过git去管理结构版本,如果真有这种场景,也可以应对。

大事务binlog

当一个事物产生的binlog量非常大的时候,比如迁移日表数据,maxwell为了控制内存使用,会自动将处理不过来的binlog放到文件系统

Using kafka version: 0.11.0.121:16:07,109 WARN  MaxwellMetrics - Metrics will not be exposed: metricsReportingType not configured.21:16:07,380 INFO  SchemaStoreSchema - Creating maxwell database21:16:07,540 INFO  Maxwell - Maxwell v?? is booting (RabbitmqProducer), starting at Position[BinlogPosition[mysql-bin.006235:24980714],lastHeartbeat=0]21:16:07,649 INFO  AbstractSchemaStore - Maxwell is capturing initial schema21:16:08,267 INFO  BinlogConnectorReplicator - Setting initial binlog pos to: mysql-bin.006235:2498071421:16:08,324 INFO  BinaryLogClient - Connected to rm-xxxxxxxxxxx.mysql.rds.aliyuncs.com:3306 at mysql-bin.006235/24980714 (sid:6379, cid:9182598)21:16:08,325 INFO  BinlogConnectorLifecycleListener - Binlog connected.03:15:36,104 INFO  ListWithDiskBuffer - Overflowed in-memory buffer, spilling over into /tmp/maxwell7935334910787514257events03:17:14,880 INFO  ListWithDiskBuffer - Overflowed in-memory buffer, spilling over into /tmp/maxwell3143086481692829045events

但是遇到另外一个问题,overflow随后就出现异常 EventDataDeserializationException:Failedto deserialize data ofEventHeaderV4,当我另起一个maxwell指点之前的binlog postion开始解析,却有没有抛异常。事后的数据也表明并没有数据丢失。

问题产生的原因还不明, Causedby:java.net.SocketException:Connectionreset,感觉像读取 binlog 流的时候还没读取到完整的event,异常关闭了连接。这个问题比较顽固,github上面类似问题都没有达到明确的解决。(这也从侧面告诉我们,大表数据迁移,也要批量进行,不要一个 insertinto..select 搞定)

03:18:20,586 INFO  ListWithDiskBuffer - Overflowed in-memory buffer, spilling over into /tmp/maxwell5229190074667071141events03:19:31,289 WARN  BinlogConnectorLifecycleListener - Communication failure.com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializationException: Failed to deserialize data of EventHeaderV4{timestamp=1514920657000, eventType=WRITE_ROWS, serverId=2115082720, headerLength=19, dataLength=8155, nextPosition=520539918, flags=0}        at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:216) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:184) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:890) [mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:559) [mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:793) [mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]Caused by: java.net.SocketException: Connection reset        at java.net.SocketInputStream.read(SocketInputStream.java:210) ~[?:1.8.0_121]        at java.net.SocketInputStream.read(SocketInputStream.java:141) ~[?:1.8.0_121]        at com.github.shyiko.mysql.binlog.io.BufferedSocketInputStream.read(BufferedSocketInputStream.java:51) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readWithinBlockBoundaries(ByteArrayInputStream.java:202) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.read(ByteArrayInputStream.java:184) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readInteger(ByteArrayInputStream.java:46) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.AbstractRowsEventDataDeserializer.deserializeLong(AbstractRowsEventDataDeserializer.java:212) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.AbstractRowsEventDataDeserializer.deserializeCell(AbstractRowsEventDataDeserializer.java:150) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.AbstractRowsEventDataDeserializer.deserializeRow(AbstractRowsEventDataDeserializer.java:132) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.WriteRowsEventDataDeserializer.deserializeRows(WriteRowsEventDataDeserializer.java:64) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.WriteRowsEventDataDeserializer.deserialize(WriteRowsEventDataDeserializer.java:56) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.WriteRowsEventDataDeserializer.deserialize(WriteRowsEventDataDeserializer.java:32) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:210) ~[mysql-binlog-connector-java-0.13.0.jar:0.13.0]        ... 5 more03:19:31,514 INFO  BinlogConnectorLifecycleListener - Binlog disconnected.03:19:31,590 WARN  BinlogConnectorReplicator - replicator stopped at position: mysql-bin.006236:520531744 -- restarting03:19:31,595 INFO  BinaryLogClient - Connected to rm-xxxxxx.mysql.rds.aliyuncs.com:3306 at mysql-bin.006236/520531744 (sid:6379, cid:9220521)

tableMapCache

前面讲过,如果我只想获取某几个表的binlog变更,需要用 include_tables 来过滤,但如果mysql server上现在删了一个表t1,但我的binlog是从昨天开始读取,被删的那个表t1在maxwell启动的时候是拉取不到表结构的。然后昨天的binlog里面有 t1 的变更,因为找不到表结构给来组装成json,会抛异常。

手动在 maxwell.tables/columns 里面插入记录是可行的。但这个问题的根本是,maxwell在binlog过滤的时候,只在处理row_event的时候,而对 tableMapCache 要求binlog里面的所有表都要有。

自己(seanlook)提交了一个commit,可以在做 tableMapCache 的时候也仅要求缓存 include_dbs/tables 这些表: https://github.com/seanlook/maxwell/commit/2618b70303078bf910a1981b69943cca75ee04fb

提高消费性能

在用rabbitmq时, routing_key%db%.%table%,但某些表产生的binlog增量非常大,就会导致各队列消息量很不平均,目前因为还没做到事务xid或者thread_id级别的并发回放,所以最小队列粒度也是表,尽量单独放一个队列,其它数据量小的合在一起。

binlog

Maxwell 在 maxwell 库中维护了 binlog 的位移等信息,由于一些原因譬如 reset master;,导致 maxwell 库中的记录与实际的binlog对不上,这时将报异常,这是可以手动修正binlog位移或者直接清空/删除 maxwell 库重建

com.github.shyiko.mysql.binlog.network.ServerException: Could not find first log file name in binary log index file        at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:885)        at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:564)        at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:796)        at java.lang.Thread.run(Thread.java:748)

以及

com.github.shyiko.mysql.binlog.network.ServerException: A slave with the same server_uuid/server_id as this slave has connected to the master; the first event 'mysql-bin.000001' at 760357, the last event read from './mysql-bin.000001' at 1888540, the last byte read from './mysql-bin.000001' at 1888540.        at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:885)        at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:564)        at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:796)        at java.lang.Thread.run(Thread.java:748)

参考文档

  • Maxwell's Daemon Doc
  • 轻风博客.MySQL Binlog解析工具Maxwell 1.17.1 的安装和使用
  • Sean.自建Binlog订阅服务 —— Maxwell
  • MySQL 5.7 基于 GTID 的主从复制实践

本文分享自微信公众号 - 小旋锋(whirlysBigData)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-03-11

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

推荐阅读

  • 「网安夜校」开课啦!多门网络安全课程开启限时优惠报名

    众志成城,共抗疫情。腾讯安全联合腾讯云大学、腾讯课堂启动「网安夜校」,为大家提供限时优惠的网络安全课程。欢迎网络安全从业者和信息安全专业学生报名参加学习,快速充电提升自我。

    腾讯安全
    安全培训腾讯云大学
  • Flink源码走读(一):Flink工程目录

    导语 | Flink已经成为未来流计算趋势,目前在很多大厂已经有了大规模的使用。最近在学习Flink源码,就想把自己学习的过程分享出来,希望能帮助到志同道合的朋友。开始阅读源码,说明读者已经对flink的基本概念有一些了解,这里就不再重复介绍Flink了。本文作为学习过程的第一章,首先对Flink的工程目录做一个解读,了解了工程下各个模块的作用,才能在遇到问题时准确定位到代码,进一步学习。

    2011aad
    大数据解决方案
  • Flink源码走读(二):Flink+Kafka实现端到端Exactly Once语义

    Flink通过Checkpoint机制实现了消息对状态影响的Exactly Once语义,即每条消息只会影响Flink内部状态有且只有一次。但无法保证输出到Sink中的数据不重复。以图一所示为例,Flink APP收到Source中的A消息,将其转化为B消息输出到Sink,APP在处理完A1后做了一次Checkpoint,假设APP在处理到A4时发生错误重启,APP将会重新从A2开始消费并处理数据,就会导致B2和B3重复输出到Sink中两次。

    2011aad
    大数据解决方案Kafka
  • kubernetes系列教程(十九)使用metric-server让HPA弹性伸缩愉快运行

    kubernetes监控指标大体可以分为两类:核心监控指标和自定义指标,核心监控指标是kubernetes内置稳定可靠监控指标,早期由heapster完成,现由metric-server实现;自定义指标用于实现核心指标的扩展,能够提供更丰富的指标支持,如应用状态指标,自定义指标需要通过Aggregator和k8s api集成,当前主流通过promethues实现。

    HappyLau谈云计算
    Kubernetes容器微服务微服务架构腾讯微服务平台 TFS
  • 三分钟入坑指北 🔜 Docsify + Serverless Framework 快速创建个人博客系统

    之前由于学摄影的关系,为了提高自己的审美,顺便锻炼下自己的英文能力,翻译了不少国外艺术类的 文章。最近一直想搭一个个人博客来存放这些内容,又懒得折腾建站,遂一直搁置。

    Aceyclee
    ServerlessHTML网站GitGitHub
  • NVM作为主存上对数据库管理系统的影响

    implications of non-volatile memory as primary storage for database management systems

    yzsDBA
    存储缓存数据库数据结构SQL
  • DevOps平台架构演进

    附最新架构图https://www.processon.com/view/5cbd897de4b0bab90962c435

    我思故我在
    DevOps 解决方案微服务架构架构设计
  • 【腾讯云AI小程序大赛】中山大学作品《小耳朵天使》

    ----------------------------------------------------------------------------------

    陈华山
    小程序 · 云开发小程序语音识别文字识别对话机器人
  • Kona JDK 在腾讯大数据领域内的实践与发展

    经常听人谈到 OpenJDK,那它到底是什么呢?相信大家都听说过 Java SE、ME、EE等规范, 通常意义上对 Open JDK 的定义指:Java SE规范的一个免费和开源参考实现。

    腾小云
    JDKJavaJVM大数据Oracle
  • 公告丨腾讯安全产品更名通知

    为了更好地为政企客户的安全保驾护航,腾讯安全即日起更新旗下身份安全、网络安全、终端安全、应用安全、数据安全、业务安全、安全管理、安全服务等八类安全产品的命名,致力于打造全栈安全产品“货架”,让客户选购安全产品/服务更加便捷,更快地找到合适的安全产品,从而对自身的安全建设“对症下药”。

    腾讯安全
    DDoS 防护应用安全 MS验证码(业务安全)应用安全(移动安全)漏洞扫描服务

扫码关注云+社区

领取腾讯云代金券