Datax快速使用之牛刀小试

前言

一次我发现业务他们在用 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.List
	at 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