题记
来自Elasticsearch中文社区的问题——
mysql中表无唯一递增字段，也无唯一递增时间字段，该怎样使用logstash实现MySQL实时增量导数据到es中？
logstash和kafka_connector都仅支持基于自增id或时间戳更新的方式增量同步数据。
回到问题本身：如果库表里没有相关字段，该怎么处理呢？
本文给出相关探讨和解决方案。
1、 binlog认知
1.1 啥是 binlog？
binlog是Mysql sever层保护的一种二进制日志，与innodb引擎中的redo/undo log是完全区别的日志；
其主要是用来记录对mysql数据更新或潜伏产生更新的SQL语句，并以"事务"的情势保存在磁盘中；
作用主要有：
1）复制：到达master-slave数据一致的目的。2）数据恢复：通过mysqlbinlog工具恢复数据。3）增量备份。
1.2 阿里的Canal实现了增量Mysql同步
一图胜千言，canal是用java开发的基于数据库增量日志解析、提供增量数据定阅&消费的中间件。
目前，canal主要支持了MySQL的binlog解析，解析完成后才利用canal client 用来处理取得的相关数据。目的：增量数据定阅&消费。
综上，使用binlog可以突破logstash或kafka-connector没有自增id或没有时间戳字段的限制，实现增量同步。
2、基于binlog的同步方式
1）基于kafka Connect的Debezium 开源工程，地址：. https://debezium.io/
2）不依赖第三方的独立利用： Maxwell开源项目，地址：http://maxwells-daemon.io/
由于已部署过conluent（kafka的企业版本，自带zookeeper、kafka、ksql、kafka-connector等），本文仅针对Debezium展开。
3、Debezium介绍
Debezium是捕获数据实时动态变化的开源的散布式同步平台。能实时捕获到数据源（Mysql、Mongo、PostgreSql）的：新增（inserts）、更新(updates)、删除(deletes)操作，实时同步到Kafka，稳定性强且速度非常快。
特点：
1）简单。无需修改利用程序。可对外提供服务。
2）稳定。延续跟踪每行的每处变动。
3）快速。构建于kafka之上，可扩大，经官方验证可处理大容量的数据。
4、同步架构
如图，Mysql到ES的同步策略，采取“曲线救国”机制。
步骤1： 基Debezium的binlog机制，将Mysql数据同步到Kafka。
步骤2： 基于Kafka_connector机制，将kafka数据同步到Elasticsearch。
5、Debezium实现Mysql到ES增删改实时同步
软件版本：
confluent：5.1.2；
Debezium：0.9.2_Final;
Mysql：5.7.x.
Elasticsearch：6.6.1
5.1 Debezium安装
confluent的安装部署参见：http://t.cn/Ef5poZk，不再赘述。
Debezium的安装只需要把debezium-connector-mysql的紧缩包解压放到Confluent的解压后的插件目录(share/java)中。
MySQL Connector plugin 紧缩包的下载地址：https://debezium.io/docs/install/。
注意重启一下confluent，以使得Debezium生效。
5.2 Mysql binlog等相关配置。
Debezium使用MySQL的binlog机制实现数据动态变化监测，所以需要Mysql提早配置binlog。
核心配置以下，在Mysql机器的/etc/my.cnf的mysqld下添加以下配置。
[mysqld]
server-id = 223344
log_bin = mysql-bin
binlog_format = row
binlog_row_image = full
expire_logs_days = 10
然后，重启一下Mysql以使得binlog生效。
systemctl start mysqld.service
5.3 配置connector连接器。
配置confluent路径目录 : /etc
创建文件夹命令 :
mkdir kafka-connect-debezium
在mysql2kafka_debezium.json寄存connector的配置信息 :
[root@localhost kafka-connect-debezium]# cat mysql2kafka_debezium.json
{
“name” : “debezium-mysql-source-0223”,
“config”:
{
“connector.class” : “io.debezium.connector.mysql.MySqlConnector”,
“database.hostname” : “192.168.1.22”,
“database.port” : “3306”,
“database.user” : “root”,
“database.password” : “XXXXXX”,
“database.whitelist” : “kafka_base_db”,
“table.whitlelist” : “accounts”,
“database.server.id” : “223344”,
“database.server.name” : “full”,
“database.history.kafka.bootstrap.servers” : “192.168.1.22:9092”,
“database.history.kafka.topic” : “account_topic”,
“include.schema.changes” : “true” ,
“incrementing.column.name” : “id”,
“database.history.skip.unparseable.ddl” : “true”,
“transforms”: “unwrap,changetopic”,
“transforms.unwrap.type”: “io.debezium.transforms.UnwrapFromEnvelope”,
“transforms.changetopic.type”:”org.apache.kafka.connect.transforms.RegexRouter”,
“transforms.changetopic.regex”:”(.*)”,
“transforms.changetopic.replacement”:”$1-smt”
}
}
注意以下配置：
“database.server.id”:对应Mysql中的server-id的配置。
“database.whitelist”: 待同步的Mysql数据库名。
“table.whitlelist”:待同步的Mysq表名。
“database.history.kafka.topic”：存储数据库的Shcema的记录信息，而非写入数据的topic
“database.server.name”:逻辑名称，每一个connector确保唯一，作为写入数据的kafka topic的前缀名称。
坑1：transforms相关5行配置作用是写入数据格式转换。
如果没有，输入数据会包括：before、after记录修改前对照信息和元数据信息（source，op，ts_ms等）。
这些信息在后续数据写入Elasticsearch是不需要的。（注意结合自己业务场景）。
格式转换相关原理：https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part⑶/
5.4 启动connector
<div
–data @mysql2kafka_debezium.json.json
http://192.168.1.22:18083/connectors | jq
/home/confluent⑸.1.0/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties
微信扫一扫打赏
支付宝扫一扫打赏
