前言

一次我发现业务他们在用 datax数据同步工具，我尤记得曾经 19 年使用过，并且基于当时的版本还修复了个 BUG并且做了数据同步管道的集成开发。没想到时间过的飞快，业务方基于海豚调度 2.0.6 的版本中有在使用，由于业务方还没有一个 web页面（虽然有开源的 DataX Web），一切又得从新开始，以下是拾遗的一部分，算作记录，后面也会对优秀的 seatunnel做些验证，有人做过 BenchMark,据说性能提升不止一点点。
搞了这么久的数据 ETL，现在的轮子真的是越来越好用了，开箱即用，能满足各种数据源的同步，本节还是讲 Datax的初步使用，后续有时间在出分析源码的文章。

扯了一些废话，以下是正文。

DataX是阿里巴巴开发的一款分布式离线数据同步工具，旨在实现多种数据源之间高效的数据交换。作为一个开源项目，DataX不仅支持多种数据源（如关系型数据库、NoSQL数据库、云存储等），还提供了灵活的插件架构，便于扩展和定制化开发。其核心功能包括数据的高效传输、增量和全量数据同步、断点续传等，这使得DataX在大数据处理中扮演了重要角色。

目前而言，结合DataX与海豚调度，可以构建一个高效、稳定的数据同步与调度平台。在实际应用中，用户可以利用DataX完成数据的抽取、转换和加载（ETL），并通过海豚调度对这些ETL任务进行统一管理和调度。这样不仅提高了数据同步的效率，还增强了数据处理流程的可靠性和可维护性。

DataX是什么？

​ DataX 是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。
详细内容请看官方介绍

一、DataX原理

DataX的实现原理是基于以下机制来实现的，

• 插件机制： DataX采用插件机制，主要包括Reader插件（数据读取模块）和Writer插件（数据写入模块）。通过这种插件化的设计，DataX可以轻松扩展支持新的数据源。常见的Reader和Writer插件包括MySQLReader、MySQLWriter、HDFSReader、HDFSWriter等。
  

• 配置驱动： DataX使用JSON格式的配置文件来定义数据同步任务。配置文件中包含Reader和Writer的具体配置信息（如连接方式、查询语句、目标表名等），以及任务的执行规则（如并行度、错误重试等）。

• 任务编排与执行： 每次DataX启动时，会根据配置文件创建并执行一个或多个同步任务。DataX的任务执行引擎负责协调Reader和Writer插件的执行：

• Reader：读取源数据并封装成中间数据格式传递给执行引擎。

• Transform（可选）：执行数据转换的逻辑（如字段映射、数据转换等），在Reader和Writer之间传递数据。

• Writer：接收经过映射转换的数据并写入到目标数据源。
  并发与数据分片： DataX支持并发和数据分片，以提高数据同步的效率。具体方法包括：

• 并发任务：DataX可以将一个大任务拆分成多个小任务，并发执行。

• 数据分片：对于大数据量的同步，DataX会将数据按一定规则分片（比如按ID范围或时间范围分片），并生成多个数据同步子任务，每个子任务并行运行。

• 容错与重试： DataX提供了多级容错机制：

  • 任务级重试：如果任务执行失败，可以设置重试次数。
  • 记录级容错：在数据同步过程中，单条记录的异常不会导致整个任务失败，而是会记录下错误信息并继续同步其他数据。

• 监控与日志： DataX提供详细的监控和日志功能，记录同步任务的执行情况，包括数据量、任务状态、错误信息等。这些信息有助于运维人员实时了解任务的执行情况，及时发现和解决问题。

总体来说，DataX通过插件化设计、配置驱动、并发数据同步和多级容错机制，保证了数据同步的灵活性、高效性和可靠性。

二、如何使用？

在此之前，我们先创一个外部表作为映射路径：
以TEXTFILE为例：

CREATE  external TABLE IF NOT EXISTS test.crm_fund_test1 (id bigint comment  'id',uid STRING COMMENT 'UID',cust_no STRING COMMENT '客户号',data_type TINYINT COMMENT '数据类型'
) STORED as textfile
location 'hdfs://localhost:9000/tmp/fund_ext'

1. mysql2hive text格式 测试

创建同步配置文件：mysql2hive.json

如下（外部表，追加方式）：

{"job": {"setting": {"speed": {"channel": 1}},"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "root","column": ["id","uid","cust_no","data_type"],"connection": [{"jdbcUrl": "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8","table": ["crm_fund_threshold"]}]}},"writer": {"name": "hdfswriter","parameter": {"path": "/tmp/fund_ext","defaultFS": "hdfs://localhost:9000","fileType": "text","writeMode": "append","compress":"NONE","encoding": "UTF-8","fieldDelimiter": ",","column": [{"name": "id","type": "bigint"},{"name": "uid","type": "string"},{"name": "cust_no","type": "string"},{"name": "data_type","type": "tinyint"}]}}}]}
}

遇到报错如下：

2024-06-30 17:10:54.163 [job-0] ERROR JobContainer - Exception when job run
java.lang.ClassCastException: java.lang.String cannot be cast to java.util.Listat com.alibaba.datax.common.util.Configuration.getList(Configuration.java:435) ~[datax-common-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.plugin.rdbms.reader.util.OriginalConfPretreatmentUtil.dealJdbcAndTable(OriginalConfPretreatmentUtil.java:85) ~[plugin-rdbms-util-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.plugin.rdbms.reader.util.OriginalConfPretreatmentUtil.simplifyConf(OriginalConfPretreatmentUtil.java:59) ~[plugin-rdbms-util-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.plugin.rdbms.reader.util.OriginalConfPretreatmentUtil.doPretreatment(OriginalConfPretreatmentUtil.java:33) ~[plugin-rdbms-util-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader$Job.init(CommonRdbmsReader.java:55) ~[plugin-rdbms-util-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.plugin.reader.mysqlreader.MysqlReader$Job.init(MysqlReader.java:37) ~[mysqlreader-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.core.job.JobContainer.initJobReader(JobContainer.java:673) ~[datax-core-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.core.job.JobContainer.init(JobContainer.java:303) ~[datax-core-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.core.job.JobContainer.start(JobContainer.java:113) ~[datax-core-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.core.Engine.start(Engine.java:86) [datax-core-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.core.Engine.entry(Engine.java:168) [datax-core-0.0.1-SNAPSHOT.jar:na]at com.alibaba.datax.core.Engine.main(Engine.java:201) [datax-core-0.0.1-SNAPSHOT.jar:na]
2024-06-30 17:10:54.167 [job-0] INFO  StandAloneJobContainerCommunicator - Total 0 records, 0 bytes | Speed 0B/s, 0 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.000s | Percentage 0.00%

注意我们这里是从 mysql同步到 hive，怀疑是没有加MySQL 驱动导致的问题。

下载并放到datax安装目录的 lib下：
wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.33/mysql-connector-java-8.0.30.jar

再次尝试报错依旧，查看下模板文件做下对比试试：

➜  bin python3 ./datax.py -r mysqlreader -w hdfswriter{"job": {"content": [{"reader": {"name": "mysqlreader","parameter": {"setting": {"content": ["name": "mysqlreader","username": "root",}"column": [],"connection": ["setting": {}"name": "mysqlreader","username": "root","crm_fund_threshold"{"jdbcUrl": [],"table": []}],"password": "","username": "","where": ""}},"writer": {"name": "hdfswriter","parameter": {"column": [],"compress": "","defaultFS": "","fieldDelimiter": "","fileName": "","fileType": "","path": "","writeMode": ""}}}],"setting": {"speed": {"channel": ""}}}
}

仔细对比了下，应该是JdbcUrl这里出的问题, java.lang.String cannot be cast to java.util.List ,也就是这里需要的是一个数组，于是修改。

再次又遇到新的报错：

2024-06-30 17:32:05.790 [job-0] ERROR JobContainer - Exception when job run
com.alibaba.datax.common.exception.DataXException: Code:[HdfsWriter-02], Description:[您填写的参数值不合法.]. - 目前TEXT FILE仅支持GZIP、BZIP2 两种压缩, 不支持您配置的 compress 模式 : [NONE]

显然，不存在 NONE 的压缩类型，修改为 GZIP 这一次成功同步，如下：

➜  job python3  ../bin/datax.py mysql2hive.json
# 最终正确的版本：
2024-06-30 17:36:57.357 [main] INFO  Engine -
{"setting":{"speed":{"channel":1}},"content":[{"reader":{"name":"mysqlreader","parameter":{"where":"","username":"root","password":"****","column":["id","uid","cust_no","data_type"],"connection":[{"jdbcUrl":["jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8"],"table":["crm_fund_threshold"]}]}},"writer":{"name":"hdfswriter","parameter":{"path":"/tmp/fund_ext","defaultFS":"hdfs://localhost:9000","fileType":"text","writeMode":"append","compress":"GZIP","encoding":"UTF-8","fileName":"fund_crm_ext","fieldDelimiter":",","column":[{"name":"id","type":"bigint"},{"name":"uid","type":"string"},{"name":"cust_no","type":"string"},{"name":"data_type","type":"tinyint"}]}}}]
}2024-06-30 17:37:09.020 [job-0] INFO  JobContainer - PerfTrace not enable!
2024-06-30 17:37:09.021 [job-0] INFO  StandAloneJobContainerCommunicator - Total 1 records, 14 bytes | Speed 1B/s, 0 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.000s | Percentage 100.00%
2024-06-30 17:37:09.022 [job-0] INFO  JobContainer -
任务启动时刻                    : 2024-06-30 17:36:57
任务结束时刻                    : 2024-06-30 17:37:09
任务总计耗时                    :                 11s
任务平均流量                    :                1B/s
记录写入速度                    :              0rec/s
读出记录总数                    :                   1
读写失败总数                    :                   0

不幸的是虽然同步成功了，但是结果却是 NULL，why ？

hive> select * from test.crm_fund_test1;
OK
NULL	NULL	NULL	NULL

这是因我们指定的上传文件的分割形式是逗号分割，但是建表的时候又没有指定，所以就出现了这个问题，解决很简单。

hive> CREATE  external TABLE IF NOT EXISTS test.crm_fund_test2(>     id bigint comment  'id',>     uid STRING COMMENT 'UID',>     cust_no STRING COMMENT '客户号',>     data_type TINYINT COMMENT '数据类型'> ) row format delimited fields terminated by ','> location 'hdfs://localhost:9000/tmp/fund_ext';
OK
Time taken: 0.219 seconds

再次运行同步命令测试，注意之前虽然我们删除了表重新创建，但由于是外部表并不会删除数据，又因为是追加模式，所以出现了多条同样的的数据，如下。

hive> select * from crm_fund_test2;
OK
23	22	123132132	2
23	22	123132132	2
23	22	123132132	2
Time taken: 0.12 seconds, Fetched: 3 row(s)

我这里修改为 truncate后变为了 1 条（MySQL 原表就只有一条）

hive> select * from crm_fund_test2;
OK
23	22	123132132	2
Time taken: 0.104 seconds, Fetched: 1 row(s)

且看日志，同名的文件会被覆盖：

2024-06-30 18:09:42.526 [job-0] INFO  JobContainer - DataX Writer.Job [hdfswriter] do prepare work .
2024-06-30 18:09:42.649 [job-0] INFO  HdfsWriter$Job - 由于您配置了writeMode truncate,  [/tmp/fund_ext] 下面的内容将被覆盖重写
2024-06-30 18:09:42.649 [job-0] INFO  HdfsWriter$Job - delete file [hdfs://localhost:9000/tmp/fund_ext/fund_crm_ext__7c76b99c_2322_40a8_9b30_cb0a84ac47a1.gz].
2024-06-30 18:09:42.657 [job-0] INFO  HdfsWriter$Job - delete file [hdfs://localhost:9000/tmp/fund_ext/fund_crm_ext__9be79eb4_39ae_47a5_9634_ea43a5a73aed.gz].
2024-06-30 18:09:42.659 [job-0] INFO  HdfsWriter$Job - delete file [hdfs://localhost:9000/tmp/fund_ext/fund_crm_ext__ca1346f6_2d23_4cea_ae1c_1d8f961aee8f.gz].
2024-06-30 18:09:42.660 [job-0] INFO  JobContainer - jobContainer starts to do split ...
2024-06-30 18:09:42.661 [job-0] INFO  JobContainer - Job set Channel-Number to 1 channels.

2. mysql2hive orc

代码如下（示例）：

{"job": {"setting": {"speed": {"channel": 1}},"content": [{"reader": {"name": "mysqlreader","parameter": {"where": "","username": "root","password": "root","column": ["id","uid","cust_no","data_type"],"connection": [{"jdbcUrl": ["jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8"],"table": ["crm_fund_threshold"]}]}},"writer": {"name": "hdfswriter","parameter": {"path": "/tmp/fund_ext2","defaultFS": "hdfs://localhost:9000","fileType": "orc","writeMode": "append","compress":"NONE","encoding": "UTF-8","fileName":"fund_crm_ext2","fieldDelimiter": ",","column": [{"name": "id","type": "bigint"},{"name": "uid","type": "string"},{"name": "cust_no","type": "string"},{"name": "data_type","type": "tinyint"}]}}}]}
}

测试结果：

CREATE  external TABLE IF NOT EXISTS test.crm_fund_test2(id bigint comment  'id',uid STRING COMMENT 'UID',cust_no STRING COMMENT '客户号',data_type TINYINT COMMENT '数据类型'   
) STORED as orc
location 'hdfs://localhost:9000/tmp/fund_ext2';# ORC 格式验证没有问题
hive> select * from test.crm_fund_test2;
OK
23	22	123132132	2

最后看下文件形式：

hive> dfs -ls /tmp/fund_ext;
Found 1 items
-rw-r--r--   3 mac supergroup         36 2024-06-30 18:13 /tmp/fund_ext/fund_crm_ext__81ce986b_6ff2_4bca_b044_d863447698cc.gz

从上面可见，文件存储采用了我们配置的 GZIP 压缩，并且使用了我们自定义的文件名前缀。

---

总结

本节内容的一个目标就是快速将 datax用起来，当然如果你有海豚，可以自行配置下结合起来测测看，后面我会将这块内容更新上。我尝试将 datax的 lib下放 MySQL 驱动包，基于当前最新的版本从实际测来看，真的不需要，因为 mysqlreader的插件已经包含了mysql-connector-java-5.1.47.jar，如下所示：

➜  mysqlreader ll libs
total 18288
-rw-r--r--@ 1 mac  staff   506K  8 21  2023 commons-collections-3.0.jar
-rw-r--r--@ 1 mac  staff   181K  8 21  2023 commons-io-2.4.jar
-rw-r--r--@ 1 mac  staff   403K  8 21  2023 commons-lang3-3.3.2.jar
-rw-r--r--@ 1 mac  staff   1.5M  8 21  2023 commons-math3-3.1.1.jar
-rw-r--r--@ 1 mac  staff   115K  8 21  2023 datax-common-0.0.1-SNAPSHOT.jar
-rw-r--r--@ 1 mac  staff   1.9M  8 21  2023 druid-1.0.15.jar
-rw-r--r--@ 1 mac  staff   1.7M  8 21  2023 fastjson2-2.0.23.jar
-rw-r--r--@ 1 mac  staff   913K  8 21  2023 guava-r05.jar
-rw-r--r--@ 1 mac  staff    44K  8 21  2023 hamcrest-core-1.3.jar
-rw-r--r--@ 1 mac  staff   258K  8 21  2023 logback-classic-1.0.13.jar
-rw-r--r--@ 1 mac  staff   409K  8 21  2023 logback-core-1.0.13.jar
-rw-r--r--@ 1 mac  staff   984K  8 21  2023 mysql-connector-java-5.1.47.jar
-rw-r--r--@ 1 mac  staff    93K  8 21  2023 plugin-rdbms-util-0.0.1-SNAPSHOT.jar
-rw-r--r--@ 1 mac  staff    31K  8 21  2023 slf4j-api-1.7.10.jar