1. 背景

      DataX 是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle 等)、HDFS、Hive、ODPS、HBase、FTP 等各种异构数据源之间稳定高效的数据同步功能。解决异构数据源同步问题，DataX 将复杂的网状的同步链路变成了星型数据链路，DataX 作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到 DataX，便能跟已有的数据源做到无缝数据同步。

       分布式 DataX 基于 datax 社区版开发的，支持分片分布式调度，分片容错，节点发现，增大数据交换吞吐，高可用，使用场景是线下一次性大规模数据同步*

2. 参考和术语

ETL 英文 Extract-Transform-Load 的缩写，用来描述将数据从来源端经过抽取（extract）、转换（transform）、加载（load）至目的端的过程

类热加载 利用自定义类加载器动态载入类，减少 metaspace 使用，减少进程启动时间，支持新类热载入

3. datax 规划

分两大平台，分布式 datax 和云 datax

 分布式 datax 用于一次性大规模数据同步

云 datax paas 平台，用于周期性的增量同步, 小规模数据

4. datax 原理介绍

*官方图，Transport 处是 Channel，本人觉得不太准确，应为 Transport

> 作业分解为任务，任务分组，最后调度器调度任务(组)

*作业分片和任务分组没有在高可用中

> 调度器负责分派资源执行任务(组)，TaskEecutor 执行任务

> transport 包括数据交换(exchanger)，转换(transformer)，交换数据字节数/记录(record)数的统计(channel)

5. datax 源码分析

原理源码分析点

5.1 配置组件

DataX 配置组件 Configuration 实质是 json 封装，用 path 获取 json 中的配置属性，与 spring boot 不同，spring boot 配置负责实例构建和初始化对象，DataX 的配置只是配置属性容器，对象获取 Configuration，使用 path 从中取出配置自身属性自行初始化

扩展 Configuration 是 json 封装，扩展新的配置熟悉非常容易，只要规划好可以对象间配置键不重复

另外，Configuration 也用于任务，任务组的定义

5.2 切分任务/任务分组

datax 作业分片执行，并行数据同步，提高吞吐；分片后分组，执行的单位任务组，任务切分由作业容器(JobContainer)完成

Ø 根据设定的限制，包括字节限制，记录数限制调整通道(channel)数量

Ø 资源根据通道数量分片，分片由插件的 Job 实现负责

分片结果是配置类 Configuration 集合

5.3 调度任务(组)/执行任务

任务切分分组完成后，进行调度和执行

调度  分配资源，分派任务，社区版的 datax 只开发单节点调度，StandaloneScheduler，调度比较简单，遍历任务组，ExecutorService 执行

TaskGroupContainerRunner  Runnable 实现，用于线程调起执行任务组

TaskGroupContainer  任务执行容错和策略

TaskExecutor 真正的任务执行，初始化转换器，构建 ReaderRunner/WriterRunner 并调起

ReaderRunner/WriterRunner 封装 reader/writer 读入写入，增加性能跟踪

RecordSender/RecordReceiver 数据交换实现，reader/writer 插件对接接口，reader 读取数据，RecordSender 到交换器, Writer 使用 RecordReceiver 接收数据写入目的数据资源

transport 参考 5.4 数据传输

插件 插件参考 5.5 插件框架

5.4 数据传输

数据传输包括，交换，转换，通道

交换(Exchanger)  datax 提供多个实现，实际只使用 BufferedRecordTransformerExchanger，带批量记录缓存，数据转换

通道(Channel) 通道负责 Record 缓存，等待 writer 接收；记录数，字节数统计和流量限制

转换(Transformer) 转换是链执行，即前一个转结果成为下一个转换的输入

5.5 读写插件框架

资源数量是无限的，并可动态增加，如果读写器启动时全加载，jdk 的 metaspace 会溢出，因此需要支持动态加载，同时支持动态增加数据源类型，datax 提供插件框架，扩展实现未有的数据源

Reader/Writer 资源读入/写入组件，两者都实现内部类 Job 和 Task

Job 负责资源分片

作业 Collector 收集作业消息

Task 负责资源具体读写

任务 Collector 收集脏数据，包括转换出错，读入出错和写入出错的 record；以可写入自定义消息

5.6 状态采集

状态采集 采集任务/任务组/插件 3 个层面的执行状态，并合并(merge)为作业状态，生成报告

5.7 性能统计

记录任务/任务组性能，汇集成作业性能统计

 PerfRecord 记录各个操作(PHASE)的开始和结束时间

PerTrace 收集 PerfRecord，统计和输出报告

6. 分布式 datax 设计

6.1 设计原理

设计借鉴 elastic-job 的分布式调度原理，首先介绍一下 elastic-job 的原理

 elastic-job 可以看作是分布式 quartz，单个作业节点使用 quartz 调起，作业集群使用 zookeeper 协调

elastic-job 所有服务都以命名空间作为边界，命名空间内的作业分配分片，共同完成整个作业，作业失效，触发 zookeeper 监听，失效分片重新分配，实现分片容错

Ø znode 存储 封装 zookeeper znode 操作

Ø 监听服务 监听 znode 变更，如，监听实例节点上下线，主节点下线

Ø 运行实例服务 注册作业实例

Ø 分片服务 主节点执行，即主节点负责分配作业分片，分片服务处理失效转移，misfired 分片

Ø 选主服务 选举主节点，相当于分布式锁

Ø 失效转移 监听运行实例，节点下线转移该节点分片到在线分片

Ø 触发服务 控制台或 api 调用服务，服务写入 trigger znode，实例节点监听触发作业执行

Ø zookeeper znode 设计

znode 设计是分布式核心所在，通过 zookeeper znode 监听实现分布式协调

6.2 技术架构

作业命名空间/job 这里的 job 与 datax 概念一致，即数据同步作业；作业命名空间是逻辑作业，可分成多个作业实例并行执行，job 调用 DataX 的 Engine 启动 Datax 作业；任务组作为作业分片

任务切分/分组 重写分片服务，在主节点执行；非主节点等待切分和分组完成；当运行实例变化，实例服务的监听触发，设置需分片标记，下次作业执行重新分片

提交作业/触发作业 用户通过控制台提交作业和触发作业，依赖触发服务

状态采集和聚合/性能采集和聚合社区版 datax 所有任务单节点中运行，作业的统计只需聚合节点的任务统计集合，分布式 datax 需要收集各节点任务统计，解决方案有两个：

1. 任务统计输出到 Prometheus，在 Prometheus 进行聚合统计

2. 节点的状态/性能输出到 redis，等待所有节点完成，选主合并统计

考虑到平台需要统计作业完成情况，选方案 2，实时状态使用 1 用于监控

附录

Ø 集成弹性资源

基于k8s弹性资源，按分片申请相应资源