問題導讀

1. Debezium有哪些特點？

2.Debezium如何實現Mysql到ES增刪改實時同步？

3.如何配置connector連接器？

題記

來自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同步

a01.jpg (24.4 KB, 下載次數: 1)

2021-3-4 10:51 上傳一圖勝千言，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、同步架構

a02.png (33.92 KB, 下載次數: 1)

2021-3-4 10:52 上傳如圖，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的前綴名稱。

坑一：transforms相關5行配置作用是寫入數據格式轉換。

如果沒有，輸入數據會包含：before、after記錄修改前對比信息以及元數據信息(source，op，ts_ms等)。這些信息在后續數據寫入Elasticsearch是不需要的。(注意結合自己業務場景)。格式轉換相關原理：https://www.confluent.io/blog/si ... thereabouts-part-3/

5.4 啟動connectorcurl -X POST -H "Content-Type:application/json"

--data @mysql2kafka_debezium.json.json

http://192.168.1.22:18083/connectors | jq復制代碼

5.5 驗證寫入是否成功。

查看kafka-topickafka-topics --list --zookeeper localhost:2181復制代碼

此處會看到寫入數據topic的信息。注意新寫入數據topic的格式：database.schema.table-smt 三部分組成。本示例topic名稱：full.kafka_base_db.account-smt。

消費數據驗證寫入是否正常

./kafka-avro-console-consumer --topic full.kafka_base_db.account-smt --bootstrap-server 192.168.1.22:9092 --from-beginning

至此，Debezium實現mysql同步kafka完成。

6、kafka-connector實現kafka同步Elasticsearch

6.1、Kafka-connector介紹

見官網:https://docs.confluent.io/current/connect.html

Kafka Connect是一個用于連接Kafka與外部系統(如數據庫，鍵值存儲，檢索系統索引和文件系統)的框架。連接器實現公共數據源數據(如Mysql、Mongo、Pgsql等)寫入Kafka，或者Kafka數據寫入目標數據庫，也可以自己開發連接器。

6.2、kafka到ES connector同步配置

配置路徑：/home/confluent-5.1.0/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties復制代碼

配置內容："connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",

"tasks.max": "1",

"topics": "full.kafka_base_db.account-smt",

"key.ignore": "true",

"connection.url": "http://192.168.1.22:9200",

"type.name": "_doc",

"name": "elasticsearch-sink-test"復制代碼6.3 kafka到ES啟動connector

啟動命令confluent load??elasticsearch-sink-test

-d /home/confluent-5.1.0/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties復制代碼

6.4 Kafka-connctor RESTFul API查看

Mysql2kafka，kafka2ES的connector詳情信息可以借助postman或者瀏覽器或者命令行查看。curl -X GET http://localhost:8083/connectors復制代碼

7、坑復盤。

坑2： 同步的過程中可能出現錯誤，比如：kafka topic沒法消費到數據。排解思路如下：

1)確認消費的topic是否是寫入數據的topic；

2)確認同步的過程中沒有出錯。可以借助connector如下命令查看。curl -X GET http://localhost:8083/connectors-xxx/status復制代碼

坑3： Mysql2ES出現日期格式不能識別。

是Mysql jar包的問題，解決方案：在my.cnf中配置時區信息即可。

坑4： kafka2ES，ES沒有寫入數據。

排解思路：

1)建議：先創建同topic名稱一致的索引，注意：Mapping靜態自定義，不要動態識別生成。

2)通過connetor/status排查出錯原因，一步步分析。

最新經典文章，歡迎關注公眾號

原文鏈接:https://blog.csdn.net/laoyang360/article/details/87897886

總結

以上是生活随笔為你收集整理的mysql hadoop架构,Debezium实现Mysql到Elasticsearch高效实时同步的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。