DataX在有赞大数据平台的实践

摘要：与大数据体系交互上报运行统计数据自带了运行结果的统计数据，我们希望把这些统计数据上报到元数据系统，作为的过程元数据存储下来。基于我们的开发策略，不要把有赞元数据系统的嵌入源码，而是在之外获取，截取出打印的统计信息再上报。

有赞大数据技术应用的早期，我们使用 Sqoop 作为数据同步工具，满足了 MySQL 与 Hive 之间数据同步的日常开发需求。

随着公司业务发展，数据同步的场景越来越多，主要是 MySQL、Hive 与文本文件之间的数据同步，Sqoop 已经不能完全满足我们的需求。在2017年初，我们已经无法忍受 Sqoop 给我们带来的折磨，准备改造我们的数据同步工具。当时有这么些很最痛的需求：

多次因 MySQL 变更引起的数据同步异常。MySQL 需要支持读写分离与分表分库模式，而且要兼容可能的数据库迁移、节点宕机以及主从切换

有不少异常是由表结构变更导致。MySQL 或 Hive 的表结构都可能发生变更，需要兼容多数的表结构不一致情况

MySQL 读写操作不要影响线上业务，不要触发 MySQL 的运维告警，不想天天被 DBA 喷

希望支持更多的数据源，如 HBase、ES、文本文件

作为数据平台管理员，还希望收集到更多运行细节，方便日常维护：

统计信息采集，例如运行时间、数据量、消耗资源

脏数据校验和上报

希望运行日志能接入公司的日志平台，方便监控

基于上述的数据同步需求，我们计划基于开源做改造，考察的对象主要是 DataX 和 Sqoop，它们之间的功能对比如下

DataX 主要的缺点在于单机运行，而这个可以通过调度系统规避，其他方面的功能均优于 Sqoop，最终我们选择了基于 DataX 开发。

使用 DataX 最重要的是解决分布式部署和运行问题，DataX 本身是单进程的客户端运行模式，需要考虑如何触发运行 DataX。

我们决定复用已有的离线任务调度系统，任务触发由调度系统负责，DataX 只负责数据同步。这样就复用了系统能力，避免重复开发。关于调度系统，可参考文章《大数据开发平台(Data Platform)在有赞的最佳践》

在每个数据平台的 worker 服务器，都会部署一个 DataX 客户端，运行时可同时启动多个进程，这些都由调度系统控制。

为了与已有的数据平台交互，需要做一些定制修改：

符合平台规则的状态上报，如启动/运行中/结束，运行时需上报进度，结束需上报成功失败

符合平台规则的运行日志实时上报，用于展示

统计、校验、流控等子模块的参数可从平台传入，并需要对结果做持久化

需要对异常输入做好兼容，例如 MySQL 主从切换、表结构变更

大致的运行流程是：前置配置文件转换、表结构校验 -> （输入 -> DataX 核心+业务无关的校验 -> 输出） -> 后置统计/持久化

尽量保证 DataX 专注于数据同步，尽量不隐含业务逻辑，把有赞特有的业务逻辑放到 DataX 之外，数据同步过程无法满足的需求，才去修改源码。

表结构、表命名规则、地址转换这些运行时前置校验逻辑，以及运行结果的持久化，放在元数据系统（参考《有赞数据仓库元数据系统实践》），而运行状态的监控放在调度系统。

DataX 并没有自带 Hive 的 reader 和 writer，而只有 HDFS 的 reader 和writer。我们选择在 DataX 之外封装，把 Hive 读写操作的配置文件，转换为 HDFS 读写的配置文件，另外辅助上 Hive DDL 操作。具体的，我们做了如下改造：

根据表名，拼接出 HDFS 路径。有赞的数据仓库规范里有一条，禁止使用外部表，这使得 HDFS 路径拼接变得容易。若是外部表，就需要从元数据系统获取相应的路径

Hive 的表结构获取，需要依赖元数据系统。还需对 Hive 表结构做校验，后面会详细说明

写 Hive 的配置里不会指定 Hive 的文件格式、分隔符，需要读取元数据，获取这些信息填入 HDFS 的写配置文件

支持新建不存在的 Hive 表或分区，能构建出符合数据仓库规范的建表语句

按 DataX 的设计理念，reader 和 writer 相互不用关心，但实际使用经常需要关联考虑才能避免运行出错。MySQL 加减字段，或者字段类型变更，都会导致 MySQL 和 Hive 的表结构不一致，需要避免这种不一致的运行出错。

非分区表都是全量导入，以 mysqlreader 配置为准。如果 MySQL 配置字段与 Hive 实际结构不一致，则把 Hive 表 drop 掉后重建。表重建可能带来下游 Hive SQL 出错的风险，这个靠 SQL 的定时检查规避。

Hive 表重建时，需要做 MySQL 字段转换为 Hive 类型，比如 MySQL 的 varchar 转为 Hive 的 string。这里有坑，Hive 没有无符号类型，注意 MySQL 的 int unsigned 的取值范围，需要向上转型，转为 Hive 的 bigint；同理，MySQL 的 bigint unsigned 也需要向上转型，我们根据实际业务情况大胆转为 bigint。而 Hive 的 string 是万能类型，如果不知道怎么转，用 string 是比较保险的。

Hive 分区表不能随意变更表结构，变更可能会导致旧分区数据读取异常。所以写Hive 分区表时，以 Hive 表结构为准，表结构不一致则直接报错。我们采取了如下的策略

这么做偏保守，对于无害的Hive分区表变更，其实可以大胆去做，比如int类型改bigint、orc表加字段。

有赞并没有独立运行的 MySQL 实例，都是由 RDS 中间件管理着 MySQL 集群，有读写分离和分表分库两种模式。读写 MySQL 有两种选择，通过 RDS 中间件读写，以及直接读写 MySQL 实例。

对于写 MySQL，写入的数据量一般不大，DataX 选择连 RDS，这样就不用额外考虑主从复制。 对于读 MySQL，考虑到有大量的全表同步任务，特别是凌晨离线任务高峰流量特别大，避免大流量对 RDS 中间件的冲击，DataX 选择直连到 MySQL 实例去读取数据。为了规避 MySQL 维护带来的地址变更风险，我们又做了几件事情：

元数据维护了标准的 RDS 中间件地址

主库、从库、RDS 中间件三者地址可以关联和任意转换

每次 DataX 任务启动时，获取最新的主库和从库地址

定期的 MySQL 连通性校验

与 DBA 建立协作关系，变更提前通知

读取 MySQL 时，对于读写分离，每次获取其中一个从库地址并连接；对于分表分库，我们有1024分片，就要转换出1024个从库地址，拼接出 DataX 的配置文件。

前提是有赞的 MySQL 建表规范，规定了建表必须有整型自增id主键。另一条运维规范，SQL 运行超过2s会被强行 kill 掉。

以读取 MySQL 全表为例，我们把一条全表去取的 SQL，拆分为很多条小 SQL，而每条小 SQL 只走主键 id 的聚簇索引，代码如下 select ... from table_name where id>");