一、数据同步平台

在项目开发中，经常需要将数据库数据同步到 ES、Redis 等其他平台，通过自己写代码进行同步过程中面临诸多问题，比如业务代码和同步代码高耦合，数据一致性等；一个支持多输入输出源的数据同步平台就显得很有意义。

概述

DataLink 数据平台基于当前成熟的 Canal、定时任务补偿、Dubbo 泛化调用等方案，实现异构数据平台的同步和最终一致性问题，支持集群部署。通过 DataLink 健康检查程序，可以对所有中间件和 DataLink 本身服务的异常情况（包括服务宕机等）进行补偿操作。
DataLink 当前主要支持对 Mysql 数据同步到 ElasticSearch，对 Oracle、Redis 以及其他 RDB 的输出正在规划中。

核心能力

Canal 实时同步
Canal 订阅 Mysql binlog，推送数据变更信息到 RocketMQ，监听 MQ 处理变更数据到 ElasticSearch
区间数据健康检查、自动补偿
基于 Elastic-Job 定时任务方式，MD5 比较 ES 与数据库信息，有差异则同步，同时解决所有中间件或同步程序可能的异常情况补偿
基于 RocketMQ 的自定义同步支持
接入方在没有部署 Canal Server，仅有 RocketMQ 的情况下，主动往 RocketMQ 发送数据请求同步

工作原理

详细流程

实时数据同步流程
健康检查与补偿流程

二、快速接入

部署中间件

部署 Canal Server，使用 MQ 模式，可参考 https://github.com/alibaba/canal
部署 ElasticSearch 6.x，参考 https://www.elastic.co/cn/blog/state-of-the-official-elasticsearch-java-clients
部署 Rocketmq

程序配置

spring:datasource:name: datalink-servicetype: com.alibaba.druid.pool.DruidDataSourceurl: jdbc:mysql://127.0.0.1:3306/ins_basicdata_prd?autoReconnect=true&characterEncoding=UTF8&useSSL=false&serverTimezone=UTCusername: meicloud_sscpassword: meiclouddriver-class-name: com.mysql.jdbc.Driverdubbo:application:name: datalinkregistry:address: zookeeper://127.0.0.1:2181?timeout=30000protocol: zookeepertimeout: 60000protocol:port: 21990name: dubboid: dubboconsumer:timeout: 6000rocketmq:# 业务方主动推送 MQ 变更配置（可选）biz:namesrvAddr: 127.0.0.1:9876groupName: datalinkBizConsumerGroup# canal监听的变更 MQ 配置canal:namesrvAddr: 127.0.0.1:9876groupName: datalinkCommonConsumerGroupelaticjob:zookeeper:server-lists: 127.0.0.1:2181namespace: datalinkelasticsearch:cluster-name: elasticsearchcluster-nodes: 127.0.0.1:9300

创建数据库表

CREATE TABLE `m_elastic_search_cfg` (`id` BIGINT(20) NOT NULL AUTO_INCREMENT COMMENT '主键id',`config_name` VARCHAR(64) NOT NULL COMMENT '配置名字',`match_topic` VARCHAR(64) NOT NULL COMMENT '匹配的topic',`listen_table` TEXT COMMENT '匹配的数据库和数据表',`primary_key` VARCHAR(64) DEFAULT NULL COMMENT '主键',`interface_name` VARCHAR(128) NOT NULL COMMENT '接口名字',`function_name` VARCHAR(128) NOT NULL COMMENT '方法名字',`search_class` VARCHAR(128) NOT NULL COMMENT '方法的参数',`search_key` VARCHAR(64) NOT NULL COMMENT '方法的参数的主键',`data_node_path` VARCHAR(64) DEFAULT NULL COMMENT '数据节点路径',`version` VARCHAR(64) DEFAULT NULL COMMENT '版本',`es_index` VARCHAR(64) NOT NULL COMMENT 'es的索引',`handler` VARCHAR(64) NOT NULL COMMENT '处理类',`increment_field` VARCHAR(64) DEFAULT NULL COMMENT '定时器补偿字段',`increment_table` TEXT COMMENT '定时器补偿表',`increment_date_column` VARCHAR(64) NOT NULL DEFAULT 'UPDATE_TIME' COMMENT '定时器补偿日期字段',`last_check_time` DATETIME DEFAULT NULL COMMENT '定时器最近同步日期',`create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',`update_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',PRIMARY KEY (`id`)
) ENGINE=INNODB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COMMENT='es配置表'

match_topic
配置格式为：ins_trade_prd_0，ins_trade_prd_[0-9]（监听 ins_trade_prd_0 到 ins_trade_prd_9）
listen_table
监听表配置，支持正则表达式，例如 ins_trade_prd_[0-9]+\.t_order_main.* 表示监听数据库 ins_trade_prd_0 到 ins_trade_prd_9 10 个数剧库，前缀为 t_order_main 的所有表（分表）。仅 handler 为 DML_NOTIFY_NORMAL 需要配置。
primary_key
主键或者从表的外键，即能从 canal 监听的 dml 数据中提取出用于 dubbo 查询的 searchKey 值；逗号间隔，依次提取
interface_name、function_name、search_class、search_key、data_node_path 和 version
都是配置 dubbo provider 的接口信息，其中 data_node_path 指定返回结果的数据路径 $.content（符合 jsonpath 语法，以 $ 开头）
handler
BIZ_NOTIFY_NORMAL，监听业务推送消息Handler
DML_NOTIFY_NORMAL，监听业务表数据变更Handler
increment_field、increment_table、increment_date_column
分别为定时任务健康检查和补偿相关字段。increment_table的配置规则同上面match_topic

启动应用

运行 DataLinkApplication 类启动应用程序，几种触发数据同步方式：
- Canal 监听的业务数据表发生变更
- 主动往业务 MQ 插入消息
- 等待定时任务健康检查、数据补偿

注意事项

同步 ES 的索引需要建立，Mapping 映射最好可以先建立（如果不建立，由 ES 动态创建 Mapping）
分库分表情况下，库需要在同一个 IP 下，能够互相访问

三、扩展：四种 CDC 方案比较优劣

抽取处理需要重点考虑增量抽取，也被称为变化数据捕获，简称 CDC。假设一个数据仓库系统，在每天夜里的业务低峰时间从操作型源系统抽取数据，那么增量抽取只需要过去 24 小时内发生变化的数据。变化数据捕获也是建立准实时数据仓库的关键技术。
当你能够识别并获得最近发生变化的数据时，抽取及其后面的转换、装载操作显然都会变得更高效，因为要处理的数据量会小很多。遗憾的是，很多源系统很难识别出最近变化的数据，或者必须侵入源系统才能做到。变化数据捕获是数据抽取中典型的技术挑战。
常用的变化数据捕获方法有时间戳、快照、触发器和日志四种。相信熟悉数据库的读者对这些方法都不会陌生。时间戳方法需要源系统有相应的数据列表示最后的数据变化。快照方法可以使用数据库系统自带的机制实现，如 Oracle 的物化视图技术，也可以自己实现相关逻辑，但会比较复杂。触发器是关系数据库系统具有的特性，源表上建立的触发器会在对该表执行 insert、update、delete 等语句时被触发，触发器中的逻辑用于捕获数据的变化。日志可以使用应用日志或系统日志，这种方式对源系统不具有侵入性，但需要额外的日志解析工作。
CDC 大体可以分为两种，一种是侵入式的，另一种是非侵入式的。所谓侵入式的是指 CDC 操作会给源系统带来性能的影响。只要 CDC 操作以任何一种方式对源库执行了 SQL 语句，就可以认为是侵入式的 CDC。基于时间戳的 CDC、基于触发器的 CDC、基于快照的 CDC 是侵入性的，基于日志的 CDC 是非侵入性的。
下表总结了四种CDC方案的特点。

项目	时间戳方式	快照方式	触发器方式	日志方式
能区分插入/更新	否	是	是	是
周期内，检测到多次更新	否	否	是	是
能检测到删除	否	是	是	是
不具有侵入性	否	否	否	是
支持实时	否	否	是	是
需要 DBA	否	否	是	是
不依赖数据库	是	是	否	否

说明：本篇是公司研发平台数据同步的一套解决方案，目前在公司多个项目中已落地，效果还不错，本人在此进行一个分享和记录，其中源码不对外开放，欢迎一起探讨~。

DataLink 数据同步平台相关推荐

去哪儿网数据同步平台技术演进与实践
作者介绍井显生,2019年加入去哪儿,现负责国内机票出票.退款.改签核心业务.在领域驱动设计(DDD).高并发有大量实践经验. 一.前言去哪儿网国内机票售后是为用户提供退票.改签.航班变动.行程服 ...
SAP工具箱数据同步平台(九与PO整合)
点击蓝字关注我们一前言数据同步平台是在ABAP中开发的一个数据同步工具,类似于LTRC,通过配置实现任意两个数据库的数据同步(ABAP需要配置相关的外部数据库连接). 数据同步平台的底层通过调 ...
异地多活数据同步平台
摘自<大型企业微服务架构实践与运营> 薛浩编著异地多活数据同步平台 1. 异地多活架构接入层应用可通过同步调用或异步消息实现相互的调用.通过相关的服务注册和发现机制保障寻址.路由.熔 ...
低代码在爱奇艺鹊桥数据同步平台的实践
前言为应对软件危机,诞生了软件工程,以期望其达到降低软件生产成本 .改进软件产品质量.提高软件生产率水平的目标.自上个世纪60年代以来,从模块化.面向对象设计到设计模式,从瀑布流模型到敏捷开发,de ...
基于dataX的数据同步平台搭建
前言基于Java和DataX工具实现数据同步的后台管理,包括数据同步任务的生成,任务的管理,查看任务的执行日志,解析任务的执行结果等功能. 内含一些技术实现方案.心得体会和填坑经验等干货. 阅读本文 ...
离线数据同步平台datax+报表可视化平台metabase
datax DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台,实现包括 MySQL.Oracle.SqlServer.Postgre.HDFS.Hive.ADS.HBase.TableS ...
【2017年第2期】应用驱动的大数据融合平台建设
孟祥飞, 冯景华, 赵洋, 夏梓峻国家超级计算天津中心,天津 300457 摘要:论述了大数据在信息社会发展中的核心地位和对信息技术创新的全方位驱动:重点阐述了应用驱动的大数据和超级计算.云计算融合 ...
可跨异构数据存储系统——数据集成平台使用教程
课程介绍数据集成(Data Integration)是阿里集团对外提供的可跨异构数据存储系统的.可靠.安全.低成本.可弹性扩展的数据同步平台,为20+种数据源提供不同网络环境下的离线(全量/增量)数 ...
神州优车开源业界领先的增量数据同步中间件——DataLink
项目介绍名称: DataLink['deitə liŋk] 译意: 数据链路,数据(自动)传输器语言: 纯java开发(JDK1.8+) 定位: 满足各种异构数据源之间的实时增量同步,一个分布式. ...

DataLink 数据同步平台

文章目录

一、数据同步平台

概述

核心能力

工作原理

详细流程

二、快速接入

部署中间件

程序配置

创建数据库表

启动应用

注意事项

三、扩展：四种 CDC 方案比较优劣

DataLink 数据同步平台相关推荐

最新文章

热门文章