关于mysql:Mysql数据实时增量同步工具之gomysqltransfer

@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、反对全量数据初始化

与同类工具比拟

实现原理

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，默认 false
web_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、delete

local 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"] = id
result["userName"] = userName
result["password"] = password
result["createTime"] = createTime
result["remark"] = remark
result["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、delete

local 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"] = id
result["userName"] = userName
result["password"] = password
result["createTime"] = createTime
result["remark"] = remark
result["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.yml

target: 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，具体如下表：

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 语句，会看到后果如下：

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_…