@TOC

数据实时增量同步工具之go-mysql-transfer：https://blog.csdn.net/weixin_...

Elasticsearch笔记之装置、配置、Kibana根底：https://blog.csdn.net/weixin_...

go-mysql-transfer官网手册：https://www.kancloud.cn/wj596...

GO笔记之环境装置：https://blog.csdn.net/weixin_...

技术选型：Mysql8 + go-mysql-transfer + ElasticSearch7.13

简介

go-mysql-transfer是一款MySQL数据库实时增量同步工具。须要GO环境

可能监听MySQL二进制日志(Binlog)的变动，将变更内容造成指定格局的音讯，实时发送到接收端。从而在数据库和接收端之间造成一个高性能、低提早的增量数据同步更新管道。

工作须要钻研了下阿里开源的MySQL Binlog增量订阅生产组件canal，其功能强大、运行稳固，然而有些方面不是太合乎需要，次要有如下三点：

1、须要本人编写客户端来生产canal解析到的数据

2、server-client模式，须要同时部署server和client两个组件，咱们的我的项目中有6个业务数据库要实时同步到redis，意味着要多部署12个组件，硬件和运维老本都会减少。

3、从server端到client端须要通过一次网络传输和序列化反序列化操作，而后再同步到接收端，感觉没有间接怼到接收端更高效。

前提条件

  • MySQL 服务器须要开启 row 模式的 binlog。
  • 因为要应用 mysqldump 命令，因而该过程的所在的服务器须要部署这一工具。
  • 这一工具应用 GoLang 开发，须要 Go 1.9+ 的环境进行构建。

  • 新版(7.13+)的本地es必须敞开平安模式才能够

    yml配置文件增加

    xpack.security.enabled: false

  • 可用的 MySQL、Elasticsearch 以及 Kibana 实例。权限须要大一些。

    • mysql binlog必须是ROW模式
    • 要同步的mysql数据表必须蕴含主键，否则间接疏忽，这是因为如果数据表没有主键，UPDATE和DELETE操作就会因为在ES中找不到对应的document而无奈进行同步
    • 不反对程序运行过程中批改表构造
    • 要赋予用于连贯mysql的账户RELOAD权限以及REPLICATION权限, SUPER权限
    GRANT REPLICATION SLAVE ON *.* TO 'elastic'@'IP';GRANT RELOAD ON *.* TO 'elastic'@'IP';UPDATE mysql.user SET Super_Priv='Y' WHERE user='elastic' AND host='IP'；

个性

1、简略，不依赖其它组件，一键部署
2、集成多种接收端，如：Redis、MongoDB、Elasticsearch、RocketMQ、Kafka、RabbitMQ、HTTP API等，无需编写客户端，开箱即用
3、内置丰盛的数据解析、音讯生成规定，反对模板语法
4、反对Lua脚本扩大，可解决简单逻辑，如：数据的转换、荡涤、打宽
5、集成Prometheus客户端，反对监控、告警
6、集成Web Admin监控页面
7、反对高可用集群部署
8、数据同步失败重试
9、反对全量数据初始化

与同类工具比拟

特色Canalmysql_streamgo-mysql-transferMaxwell
开发语言JavaPythonGolangJava
高可用反对反对反对反对
接收端编码定制Kafka等(MQ)Redis、MongoDB、Elasticsearch、RabbitMQ、Kafka、RocketMQ、HTTP API 等Kafka，Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件等
全量数据初始化不反对反对反对反对
数据格式编码定制Json（固定格局）Json（规定配置) 模板语法 Lua脚本JSON
性能(4-8TPS)

实现原理

1、go-mysql-transfer将本人伪装成MySQL的Slave，

2、向Master发送dump协定获取binlog，解析binlog并生成音讯

3、将生成的音讯实时、批量发送给接收端

如下图所示：

go-mysql部署运行

开启MySQL的binlog

批改app.yml

命令行运行
Windows间接运行 go-mysql-transfer.exe
Linux执行 nohup go-mysql-transfer &

监控

go-mysql-transfer反对两种监控模式，Prometheus和内置的Web Admin

相干配置：

# web admin相干配置enable_web_admin: true #是否启用web admin，默认falseweb_admin_port: 8060 #web监控端口,默认8060

间接拜访127.0.0.1:8060 能够看到监控界面同步数据到Elasticsearch

同步数据到Elasticsearch

配置文件——相干配置如下：

# app.yml#指标类型target: elasticsearch #elasticsearch连贯配置es_addrs: 127.0.0.1:9200 #连贯地址，多个用逗号分隔es_version: 7 # Elasticsearch版本，反对6和7、默认为7#es_password:  # 用户名#es_version:  # 明码

目前反对Elasticsearch6、Elasticsearch7两个版本

基于规定同步

相干配置如下：

rule:  -    schema: eseap #数据库名称    table: t_user #表名称    #order_by_column: id #排序字段，存量数据同步时不能为空    #column_lower_case: true #列名称转为小写,默认为false    #column_upper_case:false#列名称转为大写,默认为false    column_underscore_to_camel: true #列名称下划线转驼峰,默认为false    # 蕴含的列，多值逗号分隔，如：id,name,age,area_id  为空时示意蕴含全部列    #include_columns: ID,USER_NAME,PASSWORD    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空    #default_column_values: area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥    #date_formatter: yyyy-MM-dd #date类型格式化， 不填写默认yyyy-MM-dd    #datetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp类型格式化，不填写默认yyyy-MM-dd HH:mm:ss    #Elasticsearch相干    es_index: user_index #Index名称,能够为空，默认应用表(Table)名称    #es_mappings: #索引映射，能够为空，为空时依据数据类型自行推导ES推导    #  -      #      column: REMARK #数据库列名称    #    field: remark #映射后的ES字段名称    #    type: text #ES字段类型    #    analyzer: ik_smart #ES分词器，type为text此项有意义    #    #format: #日期格局，type为date此项有意义    #   -      #    column: USER_NAME #数据库列名称    #    field: account #映射后的ES字段名称    #    type: keyword #ES字段类型

规定示例

t_user表，数据如下：

示例一

应用上述配置
主动创立的Mapping，如下：

同步到Elasticsearch的数据如下:

示例二

配置如下：

rule:  -    schema: eseap #数据库名称    table: t_user #表名称    order_by_column: id #排序字段，存量数据同步时不能为空    column_lower_case: true #列名称转为小写,默认为false    #column_upper_case:false#列名称转为大写,默认为false    #column_underscore_to_camel: true #列名称下划线转驼峰,默认为false    # 蕴含的列，多值逗号分隔，如：id,name,age,area_id  为空时示意蕴含全部列    #include_columns: ID,USER_NAME,PASSWORD    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空    default_column_values: area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥    #date_formatter: yyyy-MM-dd #date类型格式化， 不填写默认yyyy-MM-dd    #datetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp类型格式化，不填写默认yyyy-MM-dd HH:mm:ss    #Elasticsearch相干    es_index: user_index #Index名称,能够为空，默认应用表(Table)名称    es_mappings: #索引映射，能够为空，为空时依据数据类型自行推导ES推导      -          column: REMARK #数据库列名称        field: remark #映射后的ES字段名称        type: text #ES字段类型        analyzer: ik_smart #ES分词器，type为text此项有意义        #format: #日期格局，type为date此项有意义      -          column: USER_NAME #数据库列名称        field: account #映射后的ES字段名称        type: keyword #ES字段类型

es_mappings配置项示意定义索引的mappings(映射关系)，不定义es_mappings则应用列类型主动创立索引的mappings(映射关系)。

创立的Mapping，如下：

同步到Elasticsearch的数据如下:

基于Lua脚本同步

应用Lua脚本能够实现更简单的数据处理逻辑，go-mysql-transfer反对Lua5.1语法

Lua示例

t_user表，数据如下：

示例一

引入Lua脚本：

rule:  -    schema: eseap #数据库名称    table: t_user #表名称    order_by_column: id #排序字段，存量数据同步时不能为空    lua_file_path: lua/t_user_es.lua   #lua脚本文件    es_index: user_index #Elasticsearch Index名称,能够为空，默认应用表(Table)名称    es_mappings: #索引映射，能够为空，为空时依据数据类型自行推导ES推导      -          field: id #映射后的ES字段名称        type: keyword #ES字段类型      -          field: userName #映射后的ES字段名称        type: keyword #ES字段类型      -          field: password #映射后的ES字段名称        type: keyword #ES字段类型      -          field: createTime #映射后的ES字段名称        type: date #ES字段类型        format: yyyy-MM-dd HH:mm:ss #日期格局，type为date此项有意义      -          field: remark #映射后的ES字段名称        type: text #ES字段类型        analyzer: ik_smart #ES分词器，type为text此项有意义      -          field: source #映射后的ES字段名称        type: keyword #ES字段类型

其中，
es_mappings 示意索引的mappings(映射关系)，不定义es_mappings则依据字段的值主动创立mappings(映射关系)。依据es_mappings 生成的mappings如下：

Lua脚本：

local ops = require("esOps") --加载elasticsearch操作模块local row = ops.rawRow()  --以后数据库的一行数据,table类型，key为列名称local action = ops.rawAction()  --以后数据库事件,包含：insert、update、deletelocal id = row["ID"] --获取ID列的值local userName = row["USER_NAME"] --获取USER_NAME列的值local password = row["PASSWORD"] --获取USER_NAME列的值local createTime = row["CREATE_TIME"] --获取CREATE_TIME列的值local remark = row["REMARK"] --获取REMARK列的值local result = {}  -- 定义一个table,作为后果集result["id"] = idresult["userName"] = userNameresult["password"] = passwordresult["createTime"] = createTimeresult["remark"] = remarkresult["source"] = "binlog" -- 数据起源if action == "insert" then -- 只监听新增事件    ops.INSERT("t_user",id,result) -- 新增，参数1为index名称，string类型；参数2为要插入的数据主键；参数3为要插入的数据，tablele类型或者json字符串end

同步到Elasticsearch的数据如下:

示例二

引入Lua脚本：

    schema: eseap #数据库名称    table: t_user #表名称    lua_file_path: lua/t_user_es2.lua   #lua脚本文件

未明确定义index名称、mappings，es会依据值主动创立一个名为t_user的index。

应用如下脚本：

local ops = require("esOps") --加载elasticsearch操作模块local row = ops.rawRow()  --以后数据库的一行数据,table类型，key为列名称local action = ops.rawAction()  --以后数据库事件,包含：insert、update、deletelocal id = row["ID"] --获取ID列的值local userName = row["USER_NAME"] --获取USER_NAME列的值local password = row["PASSWORD"] --获取USER_NAME列的值local createTime = row["CREATE_TIME"] --获取CREATE_TIME列的值local result = {}  -- 定义一个table,作为后果集result["id"] = idresult["userName"] = userNameresult["password"] = passwordresult["createTime"] = createTimeresult["remark"] = remarkresult["source"] = "binlog" -- 数据起源if action == "insert" then -- 只监听新增事件    ops.INSERT("t_user",id,result) -- 新增，参数1为index名称，string类型；参数2为要插入的数据主键；参数3为要插入的数据，tablele类型或者json字符串end

同步到Elasticsearch的数据如下:

esOps模块

提供的办法如下：

  1. INSERT: 插入操作,如：ops.INSERT(index,id,result)。参数index为索引名称，字符串类型；参数index为要插入数据的主键；参数result为要插入的数据，能够为table类型或者json字符串
  2. UPDATE: 批改操作,如：ops.UPDATE(index,id,result)。参数index为索引名称，字符串类型；参数index为要批改数据的主键；参数result为要批改的数据，能够为table类型或者json字符串
  3. DELETE: 删除操作,如：ops.DELETE(index,id)。参数index为索引名称，字符串类型；参数id为要删除的数据主键，类型不限；

同步数据到RocketMQ

RocketMQ配置

相干配置如下：

# app.ymltarget: rocketmq #指标类型#rocketmq连贯配置rocketmq_name_servers: 127.0.0.1:9876 #rocketmq命名服务地址，多个用逗号分隔#rocketmq_group_name: transfer_test_group #rocketmq group name,默认为空#rocketmq_instance_name: transfer_test_group_ins #rocketmq instance name,默认为空#rocketmq_access_key: RocketMQ #访问控制 accessKey,默认为空#rocketmq_secret_key: 12345678 #访问控制 secretKey,默认为空

基于规定同步

相干配置如下：

rule:  -    schema: eseap #数据库名称    table: t_user #表名称    #order_by_column: id #排序字段，存量数据同步时不能为空    #column_lower_case:false #列名称转为小写,默认为false    #column_upper_case:false#列名称转为大写,默认为false    column_underscore_to_camel: true #列名称下划线转驼峰,默认为false    # 蕴含的列，多值逗号分隔，如：id,name,age,area_id  为空时示意蕴含全部列    #include_columns: ID,USER_NAME,PASSWORD    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空    #column_mappings: CARD_NO=sfz    #列名称映射，多个映射关系用逗号分隔，如：USER_NAME=account 示意将字段名USER_NAME映射为account    #default_column_values: source=binlog,area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥    #date_formatter: yyyy-MM-dd #date类型格式化， 不填写默认yyyy-MM-dd    #datetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp类型格式化，不填写默认yyyy-MM-dd HH:mm:ss    value_encoder: json  #值编码，反对json、kv-commas、v-commas；默认为json    #value_formatter: '{{.ID}}|{{.USER_NAME}}|{{.REAL_NAME}}|{{if eq .STATUS 0}}停用{{else}}启用{{end}}'        #rocketmq相干    rocketmq_topic: transfer_test_topic #rocketmq topic，能够为空，默认应用表名称    #reserve_raw_data: false #保留update之前的数据，针对rocketmq、kafka、rabbitmq有用;默认为false

其中,
value_encoder示意值编码格局，反对json、kv-commas、v-commas三种格局，不填写默认为json，具体如下表：

格局阐明举例
jsonjson{"id": "1001","userName": "admin","password": "123456", "createTime": "2020-07-20 14:29:19"}
kv-commaskey-value逗号分隔id=1001,userName=admin,password=123456,createTime=2020-07-20 14:29:19
v-commasvalue逗号分隔1001,admin,123456,2020-07-20 14:29:19

value_formatter示意值的格式化表达式，具体模板语法参见"表达式模板"章节，当value_formatter不为空时value_encoder有效。

reserve_raw_data示意是否保留update之前的数据，即保留批改之前的老数据，默认不保留

示例

t_user表，数据如下：

在RocketMQ中创立名称为transfer_test_topic的topic，留神topic名称肯定要和rule规定中rocketmq_topic配置项的值统一

示例一

应用上述配置

insert事件，同步到RocketMQ的数据如下:

update事件，同步到RocketMQ的数据如下:

reserve_raw_data设置为true,update事件，同步到RocketMQ的数据如下:

其中，raw属性为update之前的旧数据
delete事件，同步到RocketMQ的数据如下:

示例二

配置如下：

rule:  -    schema: eseap #数据库名称    table: t_user #表名称    #order_by_column: id #排序字段，存量数据同步时不能为空    column_lower_case: true #列名称转为小写,默认为false    #column_upper_case:false#列名称转为大写,默认为false    #column_underscore_to_camel: true #列名称下划线转驼峰,默认为false    # 蕴含的列，多值逗号分隔，如：id,name,age,area_id  为空时示意蕴含全部列    #include_columns: ID,USER_NAME,PASSWORD    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空    column_mappings: USER_NAME=account    #列名称映射，多个映射关系用逗号分隔，如：USER_NAME=account 示意将字段名USER_NAME映射为account    default_column_values: area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥    #date_formatter: yyyy-MM-dd #date类型格式化， 不填写默认yyyy-MM-dd    #datetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp类型格式化，不填写默认yyyy-MM-dd HH:mm:ss    value_encoder: json  #值编码，反对json、kv-commas、v-commas；默认为json    #value_formatter: '{{.ID}}|{{.USER_NAME}}|{{.REAL_NAME}}|{{if eq .STATUS 0}}停用{{else}}启用{{end}}'    #rocketmq相干    rocketmq_topic: transfer_test_topic #rocketmq topic，能够为空，默认应用表名称    #reserve_raw_data: false #保留update之前的数据，针对rocketmq、kafka、rabbitmq有用;默认为false

其中，
column_mappings示意对列名称进行从新映射

insert事件，同步到RocketMQ的数据如下:

其中，属性名称USER_NAME变为了account

示例三

配置如下：

rule:  -    schema: eseap #数据库名称    table: t_user #表名称    #order_by_column: id #排序字段，存量数据同步时不能为空    column_lower_case: true #列名称转为小写,默认为false    #column_upper_case:false#列名称转为大写,默认为false    #column_underscore_to_camel: true #列名称下划线转驼峰,默认为false    # 蕴含的列，多值逗号分隔，如：id,name,age,area_id  为空时示意蕴含全部列    #include_columns: ID,USER_NAME,PASSWORD    #exclude_columns: BIRTHDAY,MOBIE # 排除掉的列，多值逗号分隔，如：id,name,age,area_id  默认为空    #column_mappings: USER_NAME=account    #列名称映射，多个映射关系用逗号分隔，如：USER_NAME=account 示意将字段名USER_NAME映射为account    default_column_values: area_name=合肥  #默认的列-值，多个用逗号分隔，如：source=binlog,area_name=合肥    #date_formatter: yyyy-MM-dd #date类型格式化， 不填写默认yyyy-MM-dd    #datetime_formatter: yyyy-MM-dd HH:mm:ss #datetime、timestamp类型格式化，不填写默认yyyy-MM-dd HH:mm:ss    value_encoder: v-commas  #值编码，反对json、kv-commas、v-commas；默认为json    #value_formatter: '{{.ID}}|{{.USER_NAME}}' # 值格式化表达式，如：{{.ID}}|{{.USER_NAME}},{{.ID}}示意ID字段的值、{{.USER_NAME}}示意USER_NAME字段的值    #rocketmq相干    rocketmq_topic: transfer_test_topic #rocketmq topic，能够为空，默认应用表名称

其中，
value_encoder示意音讯编码方式

insert事件，同步到RocketMQ的数据如下:

同步数据到Redis

  • Redis配置
  • 基于规定同步
  • 基于Lua脚本同步

同步数据到MongoDB

  • MongoDB配置
  • 基于规定同步
  • 基于Lua脚本同步

同步数据到RocketMQ

  • RocketMQ配置
  • 基于规定同步
  • 基于Lua脚本同步

同步数据到Kafka

  • Kafka配置
  • 基于规定同步
  • 基于Lua脚本同步

同步数据到RabbitMQ

  • RabbitMQ配置
  • 基于规定同步
  • 基于Lua脚本同步

全量数据导入

全量数据导入

Lua脚本

Lua 是一种轻量玲珑的脚本语言， 其设计目标是为了嵌入应用程序中，从而为应用程序提供灵便的扩大和定制性能。开发者只须要破费大量工夫就能大抵把握Lua的语法，照虎画猫写出可用的脚本。

基于Lua的高扩展性，能够实现更为简单的数据解析、音讯生成逻辑，定制须要的数据格式。

  • 根底模块
  • Json模块
  • HttpClient模块
  • DBClient模块

性能

性能测试地址

总结

  • go-mysql-elasticsearch 实现增量|全量 数据同步
  • go-mysql-elasticsearch能够实现同步insert、update、delete操作
  • go-mysql-elasticsearch 稳定性差点，呈现过无奈同步胜利的状况，没有具体日志，不便于排查

常见问题

如何重置同步地位(Position)

1、停掉go-mysql-transfer利用
2、在数据库执行 show master status语句，会看到后果如下：

FilePositionBinlog-Do-DBBinlog-Ignore-DB
mysql-bin.000025993779648

3、应用File和Position列的值
执行命令： ./go-mysql-transfer -config app.yml -position mysql-bin.000025 993779648

4、重启利用： ./go-mysql-transfer -config app.yml

如何同步多张表

应用yml的数组语法：

#一组连词线结尾的行，形成一个数组animal：  - Cat  - Dog  - Goldfish

go-mysql-transfer反对单库多表，也反对多库多表，配置如下：

rule:  -    schema: eseap #数据库名称    table: t_user #表名称    column_underscore_to_camel: true     value_encoder: json      redis_structure: string     redis_key_prefix: USER_  -    schema: eseap #数据库名称    table: t_sign #表名称    column_underscore_to_camel: true     value_encoder: json      redis_structure: string     redis_key_prefix: SIGN_  -    schema: gojob #数据库名称    table: t_triggered #表名称    column_underscore_to_camel: true     value_encoder: json      redis_structure: string     redis_key_prefix: TRIGGERED_

t_user表和t_sign表属于eseap数据库，t_triggered表属于gojob数据库

参考：

官网手册：https://www.kancloud.cn/wj596...

https://blog.csdn.net/weixin_...