DataX简介及使用

一、DataX离线同步工具DataX3.0介绍

1.1、 DataX 3.0概览

1.2、特征

1.3、DataX3.0框架设计

1.4、支持的数据元

1.5、DataX3.0核心架构

1.6、DataX 3.0六大核心优势

1.6.1、可靠的数据质量监控

1.6.2、丰富的数据转换功能

1.6.3、精准的速度控制

1.6.4、强劲的同步性能

1.6.5、健壮的容错机制

1.6.6、线程内部重试：

1.6.7、极简的使用体验：

二、安装

1.1、安装环境要求

1.2、安装

1.3、简单使用

三、使用DataX读取Oracle数据库同步到HDFS

3.1、官网模板：

3.1.1、模板内容说明

3.1.1.1、reader

3.1.1.2、writer

3.2、实际使用案例

参考文章：DataX/introduction.md at master · alibaba/DataX · GitHub

一、DataX离线同步工具DataX3.0介绍

DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS 等各种异构数据源之间高效的数据同步功能。

Github地址：https://github.com/alibaba/DataX

1.1、 DataX 3.0概览

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

1.2、特征

DataX本身作为数据同步框架，将不同数据源的同步抽象为从源头数据源读取数据的Reader插件，以及向目标端写入数据的Writer插件，理论上DataX框架可以支持任意数据源类型的数据同步工作。同时DataX插件体系作为一套生态系统, 每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。

1.3、DataX3.0框架设计

DataX本身作为离线数据同步框架，采用Freamwork + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。
Reader：为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
Writer：为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
Framework:：用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。

1.4、支持的数据元

类型	数据源	Reader(读)	Writer(写)	文档
RDBMS 关系型数据库	MySQL	√	√	读、写
	Oracle	√	√	读、写
	OceanBase	√	√	读、写
	SQLServer	√	√	读、写
	PostgreSQL	√	√	读、写
	DRDS	√	√	读、写
	达梦	√	√	读、写
	通用RDBMS(支持所有关系型数据库)	√	√	读、写
阿里云数仓数据存储	ODPS	√	√	读、写
	ADS		√	写
	OSS	√	√	读、写
	OCS	√	√	读、写
NoSQL数据存储	OTS	√	√	读、写
	Hbase0.94	√	√	读、写
	Hbase1.1	√	√	读、写
	MongoDB	√	√	读、写
	Hive	√	√	读、写
无结构化数据存储	TxtFile	√	√	读、写
	FTP	√	√	读、写
	HDFS	√	√	读、写
	Elasticsearch		√	写

1.5、DataX3.0核心架构

核心模块介绍：
1、DataX完成单个数据同步的作业，我们称之为Job，DataX接受到一个Job之后，将启动一个进程来完成整个作业同步过程。DataX Job模块是单个作业的中枢管理节点，承担了数据清理、子任务切分(将单一作业计算转化为多个子Task)、TaskGroup管理等功能。
2、DataXJob启动后，会根据不同的源端切分策略，将Job切分成多个小的Task(子任务)，以便于并发执行。Task便是DataX作业的最小单元，每一个Task都会负责一部分数据的同步工作。
3、切分多个Task之后，DataX Job会调用Scheduler模块，根据配置的并发数据量，将拆分成的Task重新组合，组装成TaskGroup(任务组)。每一个TaskGroup负责以一定的并发运行完毕分配好的所有Task，默认单个任务组的并发数量为5。
4、每一个Task都由TaskGroup负责启动，Task启动后，会固定启动Reader—>Channel—>Writer的线程来完成任务同步工作。
5、DataX作业运行起来之后， Job监控并等待多个TaskGroup模块任务完成，等待所有TaskGroup任务完成后Job成功退出。否则，异常退出，进程退出值非0。

DataX调度流程：
举例来说，用户提交了一个DataX作业，并且配置了20个并发，目的是将一个100张分表的mysql数据同步到odps里面。 DataX的调度决策思路是：
1、DataXJob根据分库分表切分成了100个Task。
2、根据20个并发，DataX计算共需要分配4个TaskGroup。
3、4个TaskGroup平分切分好的100个Task，每一个TaskGroup负责以5个并发共计运行25个Task。

1.6、DataX 3.0六大核心优势

1.6.1、可靠的数据质量监控

1、完美解决数据传输个别类型失真问题：

DataX旧版对于部分数据类型(比如时间戳)传输一直存在毫秒阶段等数据失真情况，新版本DataX3.0已经做到支持所有的强数据类型，每一种插件都有自己的数据类型转换策略，让数据可以完整无损的传输到目的端。

2、提供作业全链路的流量、数据量运行时监控：

DataX3.0运行过程中可以将作业本身状态、数据流量、数据速度、执行进度等信息进行全面的展示，让用户可以实时了解作业状态。并可在作业执行过程中智能判断源端和目的端的速度对比情况，给予用户更多性能排查信息。

3、提供脏数据探测：

在大量数据的传输过程中，必定会由于各种原因导致很多数据传输报错(比如类型转换错误)，这种数据DataX认为就是脏数据。DataX目前可以实现脏数据精确过滤、识别、采集、展示，为用户提供多种的脏数据处理模式，让用户准确把控数据质量大关！

1.6.2、丰富的数据转换功能

DataX作为一个服务于大数据的ETL工具，除了提供数据快照搬迁功能之外，还提供了丰富数据转换的功能，让数据在传输过程中可以轻松完成数据脱敏，补全，过滤等数据转换功能，另外还提供了自动groovy函数，让用户自定义转换函数。详情请看DataX3的transformer详细介绍。

1.6.3、精准的速度控制

还在为同步过程对在线存储压力影响而担心吗？新版本DataX3.0提供了包括通道(并发)、记录流、字节流三种流控模式，可以随意控制你的作业速度，让你的作业在库可以承受的范围内达到最佳的同步速度。

"speed": {"channel": 5,"byte": 1048576,"record": 10000
}

1.6.4、强劲的同步性能

DataX3.0每一种读插件都有一种或多种切分策略，都能将作业合理切分成多个Task并行执行，单机多线程执行模型可以让DataX速度随并发成线性增长。在源端和目的端性能都足够的情况下，单个作业一定可以打满网卡。

1.6.5、健壮的容错机制

DataX作业是极易受外部因素的干扰，网络闪断、数据源不稳定等因素很容易让同步到一半的作业报错停止。因此稳定性是DataX的基本要求，在DataX 3.0的设计中，重点完善了框架和插件的稳定性。目前DataX3.0可以做到线程级别、进程级别(暂时未开放)、作业级别多层次局部/全局的重试，保证用户的作业稳定运行。

1.6.6、线程内部重试：

DataX的核心插件都经过团队的全盘review，不同的网络交互方式都有不同的重试策略。

1、线程内部重试

DataX的核心插件都经过团队的全盘review，不同的网络交互方式都有不同的重试策略

2、线程级别重试：

目前DataX已经可以实现TaskFailover，针对于中间失败的Task，DataX框架可以做到整个Task级别的重新调度。

1.6.7、极简的使用体验：

易用：

下载即可用，支持linux和windows，只需要短短几步骤就可以完成数据的传输。

Quick start地址：

Quick Starthttps://github.com/alibaba/DataX/wiki/Quick-Start

1.6.8、详细

DataX在运行日志中打印了大量信息，其中包括传输速度，Reader、Writer性能，进程CPU，JVM和GC情况等等。

1、传输过程中打印传输速度、进度等：

2、传输过程中会打印进程相关的CPU、JVM等

3、在任务结束之后，打印总体运行情况：

二、安装

1.1、安装环境要求

•	Linux
•	JDK(1.8以上，推荐1.8)
•	Python(推荐Python2.6.X)
•	Apache Maven 3.x (Compile DataX)

Python安装地址（2.6）：CentOS 7安装Python 2.6（与已有版本共存） - 猫不夜行 - 博客园

1.2、安装

解压即用：tar -zxvf datax.tar.gz -C /opt/module/

下载后解压至本地某个目录，进入bin目录，即可运行同步作业：

cd  {YOUR_DATAX_HOME}/bin
python datax.py {YOUR_JOB.json}

自检脚本：python {YOUR_DATAX_HOME}/bin/datax.py {YOUR_DATAX_HOME}/job/job.json

出现如下所示，证明安装成功

1.3、简单使用

根据模板配置json

#stream2stream.json
{"job": {"content": [{"reader": {"name": "streamreader","parameter": {"sliceRecordCount": 10,"column": [{"type": "long","value": "10"},{"type": "string","value": "hello，你好，世界-DataX"}]}},"writer": {"name": "streamwriter","parameter": {"encoding": "UTF-8","print": true}}}],"setting": {"speed": {"channel": 1}}}
}

启动DataX：python datax.py /home/test/dataxtest/stream2stream.json

三、使用DataX读取Oracle数据库同步到HDFS

3.1、官网模板：

{"job": {"setting": {"speed": {//设置传输速度 byte/s 尽量逼近这个速度但是不高于它.// channel 表示通道数量，byte表示通道速度，如果单通道速度1MB，配置byte为1048576表示一个channel"byte": 1048576},//出错限制"errorLimit": {//先选择record"record": 0,//百分比  1表示100%"percentage": 0.02}},"content": [{"reader": {"name": "oraclereader","parameter": {// 数据库连接用户名"username": "root",// 数据库连接密码"password": "root","column": ["id","name"],//切分主键"splitPk": "db_id","connection": [{"table": ["table"],"jdbcUrl": ["jdbc:oracle:thin:@[HOST_NAME]:PORT:[DATABASE_NAME]"]}]}},"writer": {"name": "hdfswriter","parameter": {"defaultFS": "hdfs://xxx:port","fileType": "orc","path": "/user/hive/warehouse/writerorc.db/orcfull","fileName": "xxxx","column": [{"name": "col1","type": "TINYINT"},{"name": "col2","type": "SMALLINT"},{"name": "col3","type": "INT"},...],"writeMode": "append","fieldDelimiter": "\t","compress":"NONE"}}}]}
}

3.1.1、模板内容说明

3.1.1.1、reader

* **jdbcUrl**

* 描述：描述的是到对端数据库的JDBC连接信息，使用JSON的数组描述，并支持一个库填写多个连接地址。之所以使用JSON数组描述连接信息，是因为阿里集团内部支持多个IP探测，如果配置了多个，OracleReader可以依次探测ip的可连接性，直到选择一个合法的IP。如果全部连接失败，OracleReader报错。注意，jdbcUrl必须包含在connection配置单元中。对于阿里集团外部使用情况，JSON数组填写一个JDBC连接即可。

jdbcUrl按照Oracle官方规范，并可以填写连接附件控制信息。具体请参看[Oracle官方文档](http://www.oracle.com/technetwork/database/enterprise-edition/documentation/index.html)。

* 必选：是

* 默认值：无

* **username**

* 描述：数据源的用户名

* 必选：是

* 默认值：无

* **password**

* 描述：数据源指定用户名的密码

* 必选：是

* 默认值：无

* **table**

* 描述：所选取的需要同步的表。使用JSON的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需保证多张表是同一schema结构，OracleReader不予检查表是否同一逻辑表。注意，table必须包含在connection配置单元中。

* 必选：是

* 默认值：无

* **column**

* 描述：所配置的表中需要同步的列名集合，使用JSON的数组描述字段信息。用户使用\*代表默认使用所有列配置，例如['\*']。

支持列裁剪，即列可以挑选部分列进行导出。

支持列换序，即列可以不按照表schema信息进行导出。

   支持常量配置，用户需要按照JSON格式:
   ["id", "`table`", "1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"]
   id为普通列名，\`table\`为包含保留在的列名，1为整形数字常量，'bazhen.csy'为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。

Column必须显示填写，不允许为空！

* 必选：是

* 默认值：无

* **splitPk**

* 描述：OracleReader进行数据抽取时，如果指定splitPk，表示用户希望使用splitPk代表的字段进行数据分片，DataX因此会启动并发任务进行数据同步，这样可以大大提供数据同步的效能。

推荐splitPk用户使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。

目前splitPk仅支持整形、字符串型数据切分，`不支持浮点、日期等其他类型`。如果用户指定其他非支持类型，OracleReader将报错！

splitPk如果不填写，将视作用户不对单表进行切分，OracleReader使用单通道同步全量数据。

* 必选：否

* 默认值：无

* **where**

* 描述：筛选条件，MysqlReader根据指定的column、table、where条件拼接SQL，并根据这个SQL进行数据抽取。在实际业务场景中，往往会选择当天的数据进行同步，可以将where条件指定为gmt_create > $bizdate 。注意：不可以将where条件指定为limit 10，limit不是SQL的合法where子句。

where条件可以有效地进行业务增量同步。

* 必选：否

* 默认值：无

* **querySql**

* 描述：在有些业务场景下，where这一配置项不足以描述所筛选的条件，用户可以通过该配置型来自定义筛选SQL。当用户配置了这一项之后，DataX系统就会忽略table，column这些配置型，直接使用这个配置项的内容对数据进行筛选，例如需要进行多表join后同步数据，使用select a,b from table_a join table_b on table_a.id = table_b.id

`当用户配置querySql时，OracleReader直接忽略table、column、where条件的配置`。

* 必选：否

* 默认值：无

* **fetchSize**

* 描述：该配置项定义了插件和数据库服务器端每次批量数据获取条数，该值决定了DataX和服务器端的网络交互次数，能够较大的提升数据抽取性能。

`注意，该值过大(>2048)可能造成DataX进程OOM。`。

* 必选：否

* 默认值：1024

* **session**

* 描述：控制写入数据的时间格式，时区等的配置，如果表中有时间字段，配置该值以明确告知写入 oracle 的时间格式。通常配置的参数为：NLS_DATE_FORMAT,NLS_TIME_FORMAT。其配置的值为 json 格式，例如：
```
"session": [
"alter session set NLS_DATE_FORMAT='yyyy-mm-dd hh24:mi:ss'",
"alter session set NLS_TIMESTAMP_FORMAT='yyyy-mm-dd hh24:mi:ss'",
"alter session set NLS_TIMESTAMP_TZ_FORMAT='yyyy-mm-dd hh24:mi:ss'",
"alter session set TIME_ZONE='US/Pacific'"
]
```
`(注意"是 " 的转义字符串)`。

* 必选：否

* 默认值：无

### 3.3 类型转换

目前OracleReader支持大部分Oracle类型，但也存在部分个别类型没有支持的情况，请注意检查你的类型。

下面列出OracleReader针对Oracle类型转换列表:

| DataX 内部类型| Oracle 数据类型 |
| -------- | ----- |
| Long |NUMBER,INTEGER,INT,SMALLINT|
| Double |NUMERIC,DECIMAL,FLOAT,DOUBLE PRECISION,REAL|
| String |LONG,CHAR,NCHAR,VARCHAR,VARCHAR2,NVARCHAR2,CLOB,NCLOB,CHARACTER,CHARACTER VARYING,CHAR VARYING,NATIONAL CHARACTER,NATIONAL CHAR,NATIONAL CHARACTER VARYING,NATIONAL CHAR VARYING,NCHAR VARYING |
| Date |TIMESTAMP,DATE |
| Boolean |bit, bool |
| Bytes |BLOB,BFILE,RAW,LONG RAW |

请注意:

* `除上述罗列字段类型外，其他类型均不支持`。

3.1.1.2、writer

* **defaultFS**

* 描述：Hadoop hdfs文件系统namenode节点地址。格式：hdfs://ip:端口；例如：hdfs://127.0.0.1:9000

* 必选：是

* 默认值：无

* **fileType**

* 描述：文件的类型，目前只支持用户配置为"text"或"orc"。

text表示textfile文件格式

orc表示orcfile文件格式

* 必选：是

* 默认值：无 
* **path**

* 描述：存储到Hadoop hdfs文件系统的路径信息，HdfsWriter会根据并发配置在Path目录下写入多个文件。为与hive表关联，请填写hive表在hdfs上的存储路径。例：Hive上设置的数据仓库的存储路径为：/user/hive/warehouse/ ，已建立数据库：test，表：hello；则对应的存储路径为：/user/hive/warehouse/test.db/hello

* 必选：是

* 默认值：无

* **fileName**

* 描述：HdfsWriter写入时的文件名，实际执行时会在该文件名后添加随机的后缀作为每个线程写入实际文件名。

* 必选：是

* 默认值：无 
* **column**

* 描述：写入数据的字段，不支持对部分列写入。为与hive中表关联，需要指定表中所有字段名和字段类型，其中：name指定字段名，type指定字段类型。

用户可以指定Column字段信息，配置如下：

       ```json
       "column":
[
{
"name": "userName",
"type": "string"
},
{
"name": "age",
"type": "long"
}
]
       ```

* 必选：是

* 默认值：无 
* **writeMode**

* 描述：hdfswriter写入前数据清理处理模式：

       * append，写入前不做任何处理，DataX hdfswriter直接使用filename写入，并保证文件名不冲突。
       * nonConflict，如果目录下有fileName前缀的文件，直接报错。
       * truncate，如果目录下有fileName前缀的文件，先删除后写入。

* 必选：是

* 默认值：无

* **fieldDelimiter**

* 描述：hdfswriter写入时的字段分隔符,**需要用户保证与创建的Hive表的字段分隔符一致，否则无法在Hive表中查到数据**

* 必选：是

* 默认值：无

* **compress**

* 描述：hdfs文件压缩类型，默认不填写意味着没有压缩。其中：text类型文件支持压缩类型有gzip、bzip2;orc类型文件支持的压缩类型有NONE、SNAPPY（需要用户安装SnappyCodec）。

* 必选：否

* 默认值：无压缩

* **hadoopConfig**

* 描述：hadoopConfig里可以配置与Hadoop相关的一些高级参数，比如HA的配置。

       ```json
       "hadoopConfig":{
       "dfs.nameservices": "testDfs",
       "dfs.ha.namenodes.testDfs": "namenode1,namenode2",
              "dfs.namenode.rpc-address.aliDfs.namenode1": "",
       "dfs.namenode.rpc-address.aliDfs.namenode2": "",
       "dfs.client.failover.proxy.provider.testDfs": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
       }
       ```

* 必选：否

* 默认值：无

* **encoding**

* 描述：写文件的编码配置。

* 必选：否

* 默认值：utf-8，**慎重修改**

* **haveKerberos**

* 描述：是否有Kerberos认证，默认false 

例如如果用户配置true，则配置项kerberosKeytabFilePath，kerberosPrincipal为必填。

* 必选：haveKerberos 为true必选 

* 默认值：false

* **kerberosKeytabFilePath**

* 描述：Kerberos认证 keytab文件路径，绝对路径

* 必选：否 

* 默认值：无

* **kerberosPrincipal**

* 描述：Kerberos认证Principal名，如xxxx/hadoopclient@xxx.xxx

* 必选：haveKerberos 为true必选 

* 默认值：无

### 3.3 类型转换

目前 HdfsWriter 支持大部分 Hive 类型，请注意检查你的类型。

下面列出 HdfsWriter 针对 Hive 数据类型转换列表:

| DataX 内部类型| HIVE 数据类型 |
| -------- | ----- |
| Long |TINYINT,SMALLINT,INT,BIGINT |
| Double |FLOAT,DOUBLE |
| String |STRING,VARCHAR,CHAR |
| Boolean |BOOLEAN |
| Date |DATE,TIMESTAMP |

3.2、实际使用案例

配置文件：

{"job": {"content": [{"reader": {"name": "oraclereader","parameter": {"username": "scott","password": "tiger","where": "","connection": [{"querySql": ["SELECT DEPTNO,DNAME,LOC FROM DEPT"],"jdbcUrl": ["jdbc:oracle:thin:@192.168.160.135:1521:orcl"]}]}},"writer": {"name": "hdfswriter","parameter": {"defaultFS":"hdfs://mycluster","hadoopConfig":{"dfs.nameservices":"mycluster","dfs.ha.namenodes.mycluster":"hadoop1,hadoop2,hadoop3","dfs.namenode.rpc-address.mycluster.hadoop1":"hdfs://hadoop1:8020","dfs.namenode.rpc-address.mycluster.hadoop2":"hdfs://hadoop2:8020","dfs.namenode.rpc-address.mycluster.hadoop3":"hdfs://hadoop3:8020","dfs.client.failover.proxy.provider.mycluster": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider","dfs.ha.automatic-failover.enabled.yournamespace":"true"},"fileType":"orc","path":"/user/hive/warehouse/oracle_to_hive.db/dept","fileName":"dept","column":[{"name":"DEPTNO","type":"int"},{"name":"DNAME","type":"string"},{"name":"LOC","type":"string"}],"writeMode": "truncate","fieldDelimiter":"\t","encoding": "utf-8"}}}],"setting": {"speed": {"channel": "1"}}}
}

执行：python /opt/module/datax/bin/datax.py /opt/module/datax/job/oracle_to_hive.json