随着业务体量逐步增大，跨系统数据抽取的应用场景越来越多，而对于业务开发人员，需要实现异构数据源数据的抽取和录入，除了自己写代码实现数据抽取、转换、写入，另一个选择是使用开源的数据同步工具来做。对于手动实现数据同步，一方面增大了数据投产周期（需要经历应用发布完整的流程），另一方面每个人代码实现差异，项目维护难度增大。如果每个项目引入了各版本、技术栈的第三方数据同步工具，又无法实现数据安全管控，并且多引入一个中间件，也增大项目的复杂度，降低应用可移植性。

ZDTP（ZTO data transmission platform）是一个异构数据同步平台，ZDTP屏蔽各异构数据源的差异，支持数据过滤、处理与加工，并且提供状态监控、异常告警的能力。用户在ZDTP提供的SQL编辑器，写数据同步逻辑，不用关心底层的数据结构。同时容器化部署，一方面实现同步作业资源的动态扩缩容，提升物理资源利用率，另一方面增加了系统可移植性和扩展性。

1. 需要解决的问题

2. 平台介绍

一般数据同步是用户程序直接操作底层数据，首先将数据从源数据源查出，并对数据预处理，如日期格式转换、数据打码、生成主键等，然后将数据转换为目标数据源的数据格式，调用API写入数据。ZDTP首先将物理层数据结构映射为逻辑数据源，用户面向逻辑数据源，只需要关注数据处理逻辑，这样即使将数据同步到不同的数据源类型（比如从RocketMQ到Kafka，这样的操作对于项目前期预研阶段是比较常见的），只要修改目标逻辑数据源完成数据流切换。

数据同步平台能力介绍

3. 案例

下面的例子是将mysql中的user_table数据同步到RocketMQ中的主题：user_topic。其中同步作业分为两个部分：数据模型和作业SQL

同步作业修改

在ZDTP项目成立之初，首先考虑到需要面向用户简单易用，并且具备高的同步性能和高稳定性。如何能让用户以较低的成本接入并且又有较大灵活性呢？大家在日常中想必都做过数据库（如mysql）将一张表数据迁移到另一张表的同库数据迁移，只需要写一个简单的SQL如 ：INSET INTO user_bak select * from user;，只是这种数据同步是相同数据结构的，那么是不是同样使用 SQL来实现异构数据源的数据同步？

1. 技术选型-为什么选择flink

Flink作为开源的流处理和批处理框架，旨在解决大规模数据处理中的实时计算和数据流处理问题。它提供了高效、可靠和容错的流处理引擎，可用于处理实时数据流和离线批处理作业。同时支持UDF和可扩展connector，并且封装了多种API，如Stream API、Table API，有很高的灵活性和可定制性，Flink具有如下特性：

2. 架构

为了满足高可扩展性，我们将Flink集群进行容器化部署。每个同步作业都是一个小的Flink集群，这样就实现的各同步任务资源隔离，同时具备可扩展性。如果使用本地化集群部署，需要提前部署TaskManager，并预留足够多的Slot（Slot是作业执行所需的资源的最小单元），这样在就比较难做到资源快速扩容和释放。而基于容器化的集群，在Manager层，会基于每个作业需要的配置，预先计算出作业集群的规模，在启动集群时，拉取JobManager（Master进程负责集群调度）和TaskManager（工作进程负责数据处理与流转），每个TaskManager指定了Slot数据，确保所有的Slot能够分配到任务。并且在同步完成或者任务暂停，集群资源主动释放。

ZDTP架构图

3. 如何解决异构数据同步痛点

3.1 数据格式转换

Flink主要有Source（数据抽取）、数据计算（Transformation）和Sink（数据写入）三个部分

首先Source根据源端的数据类型，来匹配对应的Source connector，如JDBC connector、Kafka connector等，每个connector会实现数据数据解析逻辑，JDBC通过SQL从数据库中查询数据源、Kafka使用Pull从集群拉取数据，然后将数据类型转换为Flink Data Type。Transformation根据SQL中的处理函数和数据映射关系，将数据加工、转换，并且和目标逻辑数据源字段、数据类型一一对应，并把处理好的数据传递到Sink。Sink接收到数据后，根据数据制定的序列化格式（如json、csv），调用目标数据结构的本地化API product.send(...)将序列化数据持久化

3.2 数据完整性

在数据同步中，最棘手的莫过于数据不一致。在进行数据过程中，由于作业调整和数据源因素，会出现同步异常的情况。发生同步异常时，首先需要记录发生故障前一刻的现场，并且在任务恢复后，从上次中断的位置继续同步数据。ZDTP利用Flink 检查点和At-least-once语义来保证数据的完整性。实现原理是同步算子定时上报各自当前的同步状态（binlog保存GTID，kafka保存partition offset），ZDTP将这些状态保存到外部存储中，在下次任务启动时，从外部存储拉取任务状态，各算子从制定状态继续同步

Checkpoint的执行流程如下：

3.3 数据版本控制

数据是具有时效性的，同一个数据在不同的时间的状态会发生变化，如汇率在不同的时间可能出现上下波动，这就出现了不同时间出现不同的数据版本。数据同步过程中，一般有两种：全量数据同步（最新数据快照）和增量数据同步（基于时间历史版本数据）。根据同步类型可以将数据表划分为普通表和版本表

1. 在源端将数据的更新数据作为事件时间

2. 设置数据时间步长，如30m，每执行一个时间窗口，抽取数据的范围：[开始时间,开始时间+时间步长)

3. 抽取的数据设置主键，和事件时间一同传递到下游算子，生成upsert事件

• 数据冲突

  数据主键唯一约束upsert，数据在目标库不存在新增，存在则更新

• 数据恢复

  1.基于checkpoint 周期性执行快照，并保存最新一次成功的状态快照，并存储在数据库

  2.在任务重启、异常恢复时，根据作业id，加载任务状态快照，设置数据开启位置

• 数据回溯

  1.回溯时间、位点（kafka、rocketmq、mysql binlog）存储在数据库

  2.任务启动根据作业id，加载作业状态，设置数据开启位置

1. 一是根据同步作业的并发要求，计算出集群的规模
2. 二是在作业运行中时，根据实际流量需要的处理能力，实现物理资源（CPU、内存）动态伸缩