elasticsearch date_MySQL数据实时增量同步到Elasticsearch
Mysql到Elasticsearch的数据同步,一般用ETL来实现,但性能并不理想,目前大部分的ETL是定时查询Mysql数据库有没有新增数据或者修改数据,如果数据量小影响不大,但如果几百万上千万的数据量性能就明显的下降很多,本文是使用Go实现的go-mysql-transfer中间件来实时监控Mysql的Binlog日志,然后同步到Elasticsearch,从实时性、性能效果都不错。
一、go-mysql-transfer
go-mysql-transfer是使用Go语言实现的MySQL数据库实时增量同步工具。能够实时监听MySQL二进制日志(binlog)的变动,将变更内容形成指定格式的消息,发送到接收端。在数据库和接收端之间形成一个高性能、低延迟的增量数据(Binlog)同步管道, 具有如下特点:
1、不依赖其它组件,一键部署
2、集成多种接收端,如:Redis、MongoDB、Elasticsearch、RabbitMQ、Kafka、RocketMQ,不需要再编写客户端,开箱即用
3、内置丰富的数据解析、消息生成规则;支持Lua脚本,以处理更复杂的数据逻辑
4、支持监控告警,集成Prometheus客户端
5、高可用集群部署
6、数据同步失败重试
7、全量数据初始化
详情及安装说明 请参见: MySQL Binlog 增量同步工具go-mysql-transfer实现详解
项目开源地址:go-mysql-transfer
二、配置
# app.ymltarget: elasticsearch #目标类型#elasticsearch连接配置es_addrs: 127.0.0.1:9200 #连接地址,多个用逗号分隔es_version: 7 # Elasticsearch版本,支持6和7、默认为7#es_password: # 用户名#es_version: # 密码
三、数据转换规则
相关配置如下:
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的数据如下:
示例二
t_user表,同实例一
使用如下配置:
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语法。
示例一
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如下:
user_index索引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的数据如下:
示例二
t_user表,同实例一
引入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模块提供的方法如下:
- INSERT: 插入操作,如:ops.INSERT(index,id,result)。参数index为索引名称,字符串类型;参数index为要插入数据的主键;参数result为要插入的数据,可以为table类型或者json字符串
- UPDATE: 修改操作,如:ops.UPDATE(index,id,result)。参数index为索引名称,字符串类型;参数index为要修改数据的主键;参数result为要修改的数据,可以为table类型或者json字符串
- DELETE: 删除操作,如:ops.DELETE(index,id)。参数index为索引名称,字符串类型;参数id为要删除的数据主键,类型不限;
文章来源:https://www.jianshu.com/p/5a9b6c4f318c
elasticsearch date_MySQL数据实时增量同步到Elasticsearch相关推荐
- 数据实时增量同步之CDC工具—Canal、mysql_stream、go-mysql-transfer、Maxwell
数据实时增量同步之CDC工具-Canal.mysql_stream.go-mysql-transfer.Maxwell 什么是CDC? CDC工具对比 实现原理: Mysql binlog 讲解: m ...
- mysql增量同步kafka_MySQL数据实时增量同步到Kafka - Flume
写在前面的话 需求,将MySQL里的数据实时增量同步到Kafka.接到活儿的时候,第一个想法就是通过读取MySQL的binlog日志,将数据写到Kafka.不过对比了一些工具,例如:Canel,Dat ...
- flink实时同步mysql_基于Canal与Flink实现数据实时增量同步(一)
点击上方蓝色字体,关注我 canal是阿里巴巴旗下的一款开源项目,纯Java开发.基于数据库增量日志解析,提供增量数据订阅&消费,目前主要支持了MySQL(也支持mariaDB). 准备 配置 ...
- 简单实现MySQL数据实时增量同步到Kafka————Maxwell
任务需求:将MySQL里的数据实时增量同步到Kafka 1.准备工作 1.1.MySQL方面:开启BinLog 1.1.1.修改my.cnf文件 vi /etc/my.cnf [mysqld] ser ...
- kettle spoon判断增量更新_使用Kettle实现数据实时增量同步--时间戳增量回滚同步...
使用Kettle实现数据实时增量同步 0. 前言 本文介绍了使用Kettle对一张业务表数据(500万条数据以上)进行实时(10秒)同步,采用了时间戳增量回滚同步的方法.关于ETL和Kettle的入门 ...
- kettle 插入更新 数据增量_使用Kettle实现数据实时增量同步
2018-09-28: 示例job已上传至github,地址见文末 0. 前言 本文介绍了使用Kettle对一张业务表数据(500万条数据以上)进行实时(10秒)同步,采用了时间戳增量回滚同步的方法. ...
- mysql同步mongodb_MySQL数据实时增量同步到MongoDB
一.go-mysql-transfer go-mysql-transfer是使用Go语言实现的MySQL数据库实时增量同步工具.能够实时监听MySQL二进制日志(binlog)的变动,将变更内容形成指 ...
- Centos7.0系统下Rsync+sersync实现多文件数据实时增量同步
前言: 一.为什么要用Rsync+sersync架构? 1.sersync是基于Inotify开发的,类似于Inotify-tools的工具 2.sersync可以记录下被监听目录中发生变化的(包括增 ...
- 通过Logstash实现mysql数据定时增量同步到ES
文章目录 前言 一.系统配置 二.同步步骤整体概览 三.logstash数据同步实战 1.新建mysql表 2.ES中新建索引 3.Logstash 管道配置 4.启动Logstash 5.测试 6. ...
最新文章
- .net core 源码解析-web app是如何启动并接收处理请求(二) kestrel的启动
- JVM经常使用的调优參数
- 【项目实战】:基于python的p2p运营商数据信息的特征挖掘
- 科研实习 | 北京大学计算机系林亦波老师芯片设计软件组招收实习生
- 【物联网】OpenWrt OpenWRT的源码下载及目录结构
- 2019蚂蚁金服面试总结(Java方向)
- 单片机小白学步系列(五) 集成电路、封装相关知识
- php取掉字符串第一位支付,php怎样去掉字符串中的第一个字符
- 我们究竟需要什么!!?
- mysql数据库设置索引 以下_如何设置MySQL数据库表的索引总长度
- 【golang-GUI开发】struct tags系统(一)
- html加载js文件失败,引入js失败的可能原因
- [转]coolfire黑客入门教程系列之(五)
- Java使用ffmpeg和mencoder实现视频转码
- Mac下安装git极简教程
- Unity3D MMORPG 背包系统如何设计
- 计算机笔记Excel,秦路天善智能EXCEL学习笔记1-文本清洗函数
- 数据挖掘的过程有哪些
- 全网最全Python操作Excel教程,赶紧收藏
- html隐藏标签属性
热门文章
- 学习Spring Boot:(九)统一异常处理
- centos php-mysql_centOS 下安装php和mysql
- python数据挖掘工程师-爬虫,python工程师,数据挖掘,机器学习。我该先以哪一个为主?...
- 多媒体计算机技术19秋作业1,东师多媒体计算机技术19春在线作业1【标准答案】.doc...
- python 系统架构_Python之优化系统架构的方案
- Spring Boot————AOP入门案例及切面优先级设置
- cass坡度土方计算案例_60度斜坡怎么计算_南方CASS土方计算方法—方格网法
- python二分查找算法_如何使用python的二分查找算法
- 730阵列卡支持多大硬盘_3分钟告诉你:OPPO Reno普通版和旗舰版的差距到底有多大...
- 互联网产品生命周期_我市扎实推进全生命周期公共法律服务产品研发