杭州网站建设开发,怎么在网站做gif,滨城区住房和城乡建设局网站,温州seo品牌优化软件题记来自Elasticsearch中文社区的问题——MySQL中表无唯一递增字段#xff0c;也无唯一递增时间字段#xff0c;该怎么使用logstash实现MySQL实时增量导数据到es中#xff1f;logstash和kafka_connector都仅支持基于自增id或者时间戳更新的方式增量同步数据。回到问题本身也无唯一递增时间字段该怎么使用logstash实现MySQL实时增量导数据到es中logstash和kafka_connector都仅支持基于自增id或者时间戳更新的方式增量同步数据。回到问题本身如果库表里没有相关字段该如何处理呢本文给出相关探讨和解决方案。1、 binlog认知1.1 啥是 binlogbinlog是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增删改实时同步软件版本confluent5.1.2Debezium0.9.2_Final;Mysql5.7.x.Elasticsearch6.6.15.1 Debezium安装Debezium的安装只需要把debezium-connector-mysql的压缩包解压放到Confluent的解压后的插件目录(share/java)中。MySQL Connector plugin 压缩包的下载地址注意重启一下confluent以使得Debezium生效。5.2 Mysql binlog等相关配置。Debezium使用MySQL的binlog机制实现数据动态变化监测所以需要Mysql提前配置binlog。核心配置如下在Mysql机器的/etc/my.cnf的mysqld下添加如下配置。1[mysqld]23server-id 2233444log_bin mysql-bin5binlog_format row6binlog_row_image full7expire_logs_days 10然后重启一下Mysql以使得binlog生效。1systemctl start mysqld.service5.3 配置connector连接器。配置confluent路径目录 : /etc创建文件夹命令 :1mkdir kafka-connect-debezium在mysql2kafka_debezium.json存放connector的配置信息 :1[rootlocalhost kafka-connect-debezium]# cat mysql2kafka_debezium.json2{3 name : debezium-mysql-source-0223,4 config:5 {6 connector.class : io.debezium.connector.mysql.MySqlConnector,7 database.hostname : 192.168.1.22,8 database.port : 3306,9 database.user : root,10 database.password : XXXXXX,11 database.whitelist : kafka_base_db,12 table.whitlelist : accounts,13 database.server.id : 223344,14 database.server.name : full,15 database.history.kafka.bootstrap.servers : 192.168.1.22:9092,16 database.history.kafka.topic : account_topic,17 include.schema.changes : true ,18 incrementing.column.name : id,19 database.history.skip.unparseable.ddl : true,20 transforms: unwrap,changetopic,21 transforms.unwrap.type: io.debezium.transforms.UnwrapFromEnvelope,22 transforms.changetopic.type:org.apache.kafka.connect.transforms.RegexRouter,23 transforms.changetopic.regex:(.*),24 transforms.changetopic.replacement:$1-smt25 }26}注意如下配置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记录修改前对比信息以及元数据信息(sourceopts_ms等)。这些信息在后续数据写入Elasticsearch是不需要的。(注意结合自己业务场景)。5.4 启动connector1curl -X POST -H Content-Type:application/json2--data mysql2kafka_debezium.json.json3http://192.168.1.22:18083/connectors | jq5.5 验证写入是否成功。5.5.1 查看kafka-topic1 kafka-topics --list --zookeeper localhost:2181此处会看到写入数据topic的信息。注意新写入数据topic的格式database.schema.table-smt 三部分组成。本示例topic名称full.kafka_base_db.account-smt5.5.2 消费数据验证写入是否正常1./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同步Elasticsearch6.1、Kafka-connector介绍Kafka Connect是一个用于连接Kafka与外部系统(如数据库键值存储检索系统索引和文件系统)的框架。连接器实现公共数据源数据(如Mysql、Mongo、Pgsql等)写入Kafka或者Kafka数据写入目标数据库也可以自己开发连接器。6.2、kafka到ES connector同步配置配置路径1/home/confluent-5.1.0/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties配置内容1connector.class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector,2tasks.max: 1,3topics: full.kafka_base_db.account-smt,4key.ignore: true,5connection.url: http://192.168.1.22:9200,6type.name: _doc,7name: elasticsearch-sink-test6.3 kafka到ES启动connector启动命令1confluent load elasticsearch-sink-test2-d /home/confluent-5.1.0/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties6.4 Kafka-connctor RESTFul API查看Mysql2kafkakafka2ES的connector详情信息可以借助postman或者浏览器或者命令行查看。1curl -X GET http://localhost:8083/connectors7、坑复盘。坑2 同步的过程中可能出现错误比如kafka topic没法消费到数据。排解思路如下1)确认消费的topic是否是写入数据的topic2)确认同步的过程中没有出错。可以借助connector如下命令查看。1curl -X GET http://localhost:8083/connectors-xxx/status坑3 Mysql2ES出现日期格式不能识别。是Mysql jar包的问题解决方案在my.cnf中配置时区信息即可。坑4 kafka2ESES没有写入数据。排解思路1)建议先创建同topic名称一致的索引注意Mapping静态自定义不要动态识别生成。2)通过connetor/status排查出错原因一步步分析。8、小结binlog的实现突破了字段的限制实际上业界的go-mysql-elasticsearch已经实现。对比logstash、kafka-connector虽然Debezium“曲线救国”两步实现了实时同步但稳定性实时性能相对不错。推荐大家使用。大家有好的同步方式也欢迎留言讨论交流。推荐阅读重磅 | 死磕Elasticsearch方法论认知清单(2019春节更新版)Elasticsearch基础、进阶、实战第一公众号
