DataX框架的设计、运行原理详解
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了DataX框架的设计、运行原理详解相关的知识，希望对你有一定的参考价值。
什么是DataX DataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(mysql、Oracle等)、HDFS、Hive、ODPS、HBa
参考技术A 什么是DataXDataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。
方法/步骤
DataX的设计
为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。
框架设计
DataX本身作为离线数据同步框架，采用Framework + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。
Reader：数据采集模块，负责采集数据源的数据，将数据发送给Framework。
Writer：数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
Framework：用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。
运行原理
① DataX完成单个数据同步的作业，我们称之为Job，DataX接受到一个Job之后，将启动一个进程来完成整个作业同步过程。DataX Job模块是单个作业的中枢管理节点，承担了数据清理、子任务切分(将单一作业计算转化为多个子Task)、TaskGroup管理等功能。
②DataXJob启动后，会根据不同的源端切分策略，将Job切分成多个小的Task(子任务)，以便于并发执行。Task便是DataX作业的最小单元，每一个Task都会负责一部分数据的同步工作。
③切分多个Task之后，DataX Job会调用Scheduler模块，根据配置的并发数据量，将拆分成的Task重新组合，组装成TaskGroup(任务组)。每一个TaskGroup负责以一定的并发运行完毕分配好的所有Task，默认单个任务组的并发数量为5。
④每一个Task都由TaskGroup负责启动，Task启动后，会固定启动Reader—>Channel—>Writer的线程来完成任务同步工作。
⑤DataX作业运行起来之后， Job监控并等待多个TaskGroup模块任务完成，等待所有TaskGroup任务完成后Job成功退出。否则，异常退出，进程退出值非0
DataX 使用详解
目录
3.2 通过 DataX 实现 MySQL 数据同步Elasticsearch
3.3 通过 DataX 实现 Mongodb数据同步MySQL
3.4 通过 DataX 实现 Oracle数据同步MySQL
一、Datax 概述
1.1 Datax
DataX 是阿里云 DataWorks数据集成 的开源版本，在阿里巴巴集团内被广泛使用的离线数据同步工具/平台。DataX 实现了包括 MySQL、Oracle、OceanBase、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、Hologres、DRDS 等各种异构数据源之间高效的数据同步功能。
1.2 Datax Features
DataX本身作为数据同步框架，将不同数据源的同步抽象为从源头数据源读取数据的Reader插件，以及向目标端写入数据的Writer插件，理论上DataX框架可以支持任意数据源类型的数据同步工作。同时DataX插件体系作为一套生态系统, 每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。
1.3 Datax 环境要求
- Linux
- JDK(1.8以上，推荐1.8)
- Python(2或3都可以)
- Apache Maven 3.x (Compile DataX)
1.4 Datax 安装
方法一、直接下载DataX工具包：DataX下载地址
下载后解压至本地某个目录，进入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
方法二、下载DataX源码，自己编译：DataX源码
(1)、下载DataX源码：
$ git clone git@github.com:alibaba/DataX.git
(2)、通过maven打包：
$ cd DataX_source_code_home $ mvn -U clean package assembly:assembly -Dmaven.test.skip=true
打包成功，日志显示如下：
[INFO] BUILD SUCCESS
[INFO] -----------------------------------------------------------------
[INFO] Total time: 08:12 min
[INFO] Finished at: 2015-12-13T16:26:48+08:00
[INFO] Final Memory: 133M/960M
[INFO] -----------------------------------------------------------------
打包成功后的DataX包位于 DataX_source_code_home/target/datax/datax/ ，结构如下：
$ cd DataX_source_code_home $ ls ./target/datax/datax/ bin conf job lib log log_perf plugin script
1.5 Datax 演示示例
功能描述：从stream读取数据并打印到控制台
第一步：创建作业的配置文件（json格式）
通过命令查看配置模板： python datax.py -r YOUR_READER -w YOUR_WRITER
温馨提示： YOUR_READER ：数据源读取类型，YOUR_WRITER: 数据源写入类型。稍后补充datax 支持数据类型和对应的read/write.
$ cd YOUR_DATAX_HOME/bin
# 温馨提示：本示例使用的streamread/streamwrite 流做功能演示。
$ python datax.py -r streamreader -w streamwriter
"job":
"content": [
"reader":
"name": "streamreader",
"parameter":
"column": [],
"sliceRecordCount": ""
,
"writer":
"name": "streamwriter",
"parameter":
"encoding": "",
"print": true
],
"setting":
"speed":
"channel": ""
根据模板配置文件，修改相关参数。
定义切片记录总数(输出总数)："sliceRecordCount": 10
定义字段："column":["type":"long"， "value":"10"，"type": "string"，"value": "hello，你好，世界-DataX"]
$ vi streamDemo.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": 5
第二步：启动DataX
$ cd YOUR_DATAX_DIR_BIN
$ python datax.py ./stream2stream.json datax 显示日志如下:
...
2022-12-16 11:20:25.263 [job-0] INFO JobContainer -
任务启动时刻 : 2015-12-17 11:20:15
任务结束时刻 : 2015-12-17 11:20:25
任务总计耗时 : 10s
任务平均流量 : 205B/s
记录写入速度 : 5rec/s
读出记录总数 : 50
读写失败总数 : 0二、Datax 核心详解
2.1 DataX 3.0概览
DataX 是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。
-
设计理念
为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。
-
当前使用现状
DataX在阿里巴巴集团内被广泛使用，承担了所有大数据的离线同步业务，并已持续稳定运行了6年之久。目前每天完成同步8w多道作业，每日传输数据量超过300TB。
此前已经开源DataX1.0版本，此次介绍为阿里云开源全新版本DataX3.0，有了更多更强大的功能和更好的使用体验。Github主页地址：GitHub - alibaba/DataX: DataX是阿里云DataWorks数据集成的开源版本。
2.2 DataX3.0框架设计
DataX本身作为离线数据同步框架，采用Framework + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。
- Reader：Reader为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
- Writer： Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
- Framework：Framework用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。
2.3 DataX3.0 支持的读写插件
经过几年积累，DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入。DataX目前支持数据如下：
| 类型 | 数据源 | 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 | √ | 写 |
2.4 DataX3.0核心架构
DataX 3.0 开源版本支持单机多线程模式完成同步作业运行，本小节按一个DataX作业生命周期的时序图，从整体架构设计非常简要说明DataX各个模块相互关系。
核心模块介绍：
- DataX完成单个数据同步的作业，我们称之为Job，DataX接受到一个Job之后，将启动一个进程来完成整个作业同步过程。DataX Job模块是单个作业的中枢管理节点，承担了数据清理、子任务切分(将单一作业计算转化为多个子Task)、TaskGroup管理等功能。
- DataXJob启动后，会根据不同的源端切分策略，将Job切分成多个小的Task(子任务)，以便于并发执行。Task便是DataX作业的最小单元，每一个Task都会负责一部分数据的同步工作。
- 切分多个Task之后，DataX Job会调用Scheduler模块，根据配置的并发数据量，将拆分成的Task重新组合，组装成TaskGroup(任务组)。每一个TaskGroup负责以一定的并发运行完毕分配好的所有Task，默认单个任务组的并发数量为5。
- 每一个Task都由TaskGroup负责启动，Task启动后，会固定启动Reader—>Channel—>Writer的线程来完成任务同步工作。
- DataX作业运行起来之后， Job监控并等待多个TaskGroup模块任务完成，等待所有TaskGroup任务完成后Job成功退出。否则，异常退出，进程退出值非0
DataX调度流程：
举例来说，用户提交了一个DataX作业，并且配置了20个并发，目的是将一个100张分表的mysql数据同步到odps里面。 DataX的调度决策思路是：
- DataXJob根据分库分表切分成了100个Task。
- 根据20个并发，DataX计算共需要分配4个TaskGroup。
- 4个TaskGroup平分切分好的100个Task，每一个TaskGroup负责以5个并发共计运行25个Task。
2.5 DataX3.0 六大核心优势
-
可靠的数据质量监控
-
完美解决数据传输个别类型失真问题
DataX旧版对于部分数据类型(比如时间戳)传输一直存在毫秒阶段等数据失真情况，新版本DataX3.0已经做到支持所有的强数据类型，每一种插件都有自己的数据类型转换策略，让数据可以完整无损的传输到目的端。
-
提供作业全链路的流量、数据量运行时监控
DataX3.0运行过程中可以将作业本身状态、数据流量、数据速度、执行进度等信息进行全面的展示，让用户可以实时了解作业状态。并可在作业执行过程中智能判断源端和目的端的速度对比情况，给予用户更多性能排查信息。
-
提供脏数据探测
在大量数据的传输过程中，必定会由于各种原因导致很多数据传输报错(比如类型转换错误)，这种数据DataX认为就是脏数据。DataX目前可以实现脏数据精确过滤、识别、采集、展示，为用户提供多种的脏数据处理模式，让用户准确把控数据质量大关！
-
-
丰富的数据转换功能
DataX作为一个服务于大数据的ETL工具，除了提供数据快照搬迁功能之外，还提供了丰富数据转换的功能，让数据在传输过程中可以轻松完成数据脱敏，补全，过滤等数据转换功能，另外还提供了自动groovy函数，让用户自定义转换函数。详情请看DataX3的transformer详细介绍。
-
精准的速度控制
还在为同步过程对在线存储压力影响而担心吗？新版本DataX3.0提供了包括通道(并发)、记录流、字节流三种流控模式，可以随意控制你的作业速度，让你的作业在库可以承受的范围内达到最佳的同步速度。
"speed": "channel": 5, "byte": 1048576, "record": 10000
-
强劲的同步性能
DataX3.0每一种读插件都有一种或多种切分策略，都能将作业合理切分成多个Task并行执行，单机多线程执行模型可以让DataX速度随并发成线性增长。在源端和目的端性能都足够的情况下，单个作业一定可以打满网卡。另外，DataX团队对所有的已经接入的插件都做了极致的性能优化，并且做了完整的性能测试。性能测试相关详情可以参照每单个数据源的详细介绍：DataX数据源指南
-
健壮的容错机制
DataX作业是极易受外部因素的干扰，网络闪断、数据源不稳定等因素很容易让同步到一半的作业报错停止。因此稳定性是DataX的基本要求，在DataX 3.0的设计中，重点完善了框架和插件的稳定性。目前DataX3.0可以做到线程级别、进程级别(暂时未开放)、作业级别多层次局部/全局的重试，保证用户的作业稳定运行。
-
线程内部重试
DataX的核心插件都经过团队的全盘review，不同的网络交互方式都有不同的重试策略。
-
线程级别重试
目前DataX已经可以实现TaskFailover，针对于中间失败的Task，DataX框架可以做到整个Task级别的重新调度。
-
-
极简的使用体验
-
易用
下载即可用，支持linux和windows，只需要短短几步骤就可以完成数据的传输。请点击：Quick Start
-
详细
DataX在运行日志中打印了大量信息，其中包括传输速度，Reader、Writer性能，进程CPU，JVM和GC情况等等。
-
传输过程中打印传输速度、进度等
-
传输过程中会打印进程相关的CPU、JVM等
-
在任务结束之后，打印总体运行情况
-
三、Datax 实现数据同步
3.1 通过 DataX 实 MySQL 数据同步
第一步：生成 MySQL 到 MySQL 数据同步的模板。
[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py -r mysqlreader -w mysqlwriter
"job":
"content": [
"reader":
"name": "mysqlreader", # 读取端
"parameter":
"column": [], # 需要同步的列 (* 表示所有的列)
"connection": [
"jdbcUrl": [], # 连接信息
"table": [] # 连接表
],
"password": "", # 连接用户
"username": "", # 连接密码
"where": "" # 描述筛选条件
,
"writer":
"name": "mysqlwriter", # 写入端
"parameter":
"column": [], # 需要同步的列
"connection": [
"jdbcUrl": "", # 连接信息
"table": [] # 连接表
],
"password": "", # 连接密码
"preSql": [], # 同步前. 要做的事
"session": [],
"username": "", # 连接用户
"writeMode": "" # 操作类型
],
"setting":
"speed":
"channel": "" # 指定并发数
第二步：编写MySQL数据同步文件:MySQLToMySQL.json
[root@MySQL-1 ~]# vi MySQLToMySQL.json
"job":
"content": [
"reader":
"name": "mysqlreader",
"parameter":
"username": "root",
"password": "123123",
"column": ["*"],
"splitPk": "ID",
"connection": [
"jdbcUrl": [
"jdbc:mysql://192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"
],
"table": ["t_member"]
]
,
"writer":
"name": "mysqlwriter",
"parameter":
"column": ["*"],
"connection": [
"jdbcUrl": "jdbc:mysql://192.168.1.2:3306/course-study?useUnicode=true&characterEncoding=utf8",
"table": ["t_member"]
],
"password": "123123",
"preSql": [
"truncate t_member"
],
"username": "root",
"writeMode": "insert"
],
"setting":
"speed":
"channel": "5"
第三步：验证
[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py MySQLToMySQL.json
控制台输出:
2021-12-15 16:45:15.120 [job-0] INFO JobContainer - PerfTrace not enable!
2021-12-15 16:45:15.120 [job-0] INFO StandAloneJobContainerCommunicator - Total 2999999 records, 107666651 bytes | Speed 2.57MB/s, 74999 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 82.173s | All Task WaitReaderTime 75.722s | Percentage 100.00%
2021-12-15 16:45:15.124 [job-0] INFO JobContainer -
任务启动时刻 : 2021-12-15 16:44:32
任务结束时刻 : 2021-12-15 16:45:15
任务总计耗时 : 42s
任务平均流量 : 2.57MB/s
记录写入速度 : 74999rec/s
读出记录总数 : 2999999
读写失败总数 : 0
3.2 通过 DataX 实现 MySQL 数据同步Elasticsearch
第一步：生成 MySQL 到 Elasticsearch 数据同步的模板。
[root@MySQL-1 ~]# python datax.py -r mysqlreader -w elasticsearchwriter
Read writer [elasticsearchwriter] template error:: can't find file /home/**/tool/datax/plugin/writer/elasticsearchwriter/plug_job_template.json
问题原因 ：写elasticsearchwriter 插件 没有生成plugin_job_template.json文件
解决办法：在/home/**/tool/datax/plugin/writer/elasticsearchwriter/文件目录下，新增plugin_job_template.json 文件，并为plugin_job_template.json添加相关权限。
vi plugin_job_template.json
# 编辑plugin_job_template.json 内容
"name"： "elasticsearchwriter",
"paramter":
"endpoint": "",
"index": "",
"type": "",
"cleanup": true,
"dynamic": true,
"settings":"index":"number_of_shards":2, "nummber_of_replicas": 1,
"batchSize": 10000,
"splitter": "",
"column": []
再次执行第一步，获取MySQL 同步Elasticsearch数据模板内容。
"job":
"content": [
"reader":
"name": "mysqlreader", # 读取端
"parameter":
"column": [], # 需要同步的列 (* 表示所有的列)
"connection": [
"jdbcUrl": [], # 连接信息
"table": [] # 连接表
],
"password": "", # 连接用户
"username": "", # 连接密码
"where": "" # 描述筛选条件
,
"writer":
"name": "elasticsearchwriter", # 写入端
"paramter":
"endpoint": "",
"index": "",
"type": "",
"cleanup": true,
"dynamic": true,
"settings":
"index":
"number_of_shards":2,
"nummber_of_replicas": 1
,
"batchSize": 10000,
"splitter": "",
"column": []
],
"setting":
"speed":
"channel": "" # 指定并发数
第二步：创建es 索引并指定mapping 映射。
PUT http://192.168.43.10:9200/es_db
"mappings":
"properties":
"address":
"type":"text",
"fields":
"keyword":
"type":"keyword",
"ignore_above":256
,
"age":
"type":"long"
,
"book":
"type":"text",
"fields":
"keyword":
"type":"keyword",
"ignore_above":256
,
"name":
"type":"text",
"fields":
"keyword":
"type":"keyword",
"ignore_above":256
,
"sex":
"type":"long"
第三步：编写MySQL数据同步文件:MySQLToES.json
[root@MySQL-1 ~]# vi MySQLToES.json
"job":
"content": [
"reader":
"name": "mysqlreader",
"parameter":
"username": "root",
"password": "123123",
"where":"created_dt>='$start_time 00:00:00' and created_dt<='$end_time 23:59:59'",
"column": [
"id",
"address",
"age",
"book",
"name",
"sex"
],
"connection": [
"jdbcUrl": [
"jdbc:mysql://192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"
],
"table": ["t_user"]
]
,
"writer":
"name": "elasticsearchwriter",
"parameter":
"endpoint": "http://127.0.0.1:9200",
"index":"es_db",
"type":"traces",
"settings": "index" :"number_of_shards": 5, "number_of_replicas": 1,
"batchSize": 5000,
"splitter": ",",
"column": [
"name":"pk","type":"id",
"name":"address","type":"keyword",
"name":"book","type":"keyword",
"name":"sex","type":"integer",
"name":"age","type":"integer",
"name":"name","type":"keyword"
]
],
"setting":
"speed":
"channel": "5"
第四步：验证
python /usr/local/datax/bin/datax.py ./MySQLToES.json -p "-Dstart_time=2022-12-17 -Dend_time=2022-12-17"3.3 通过 DataX 实现 Mongodb数据同步MySQL
第一步：生成 Mongodb 到 MySQL数据同步的模板。
[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py -r mongodbreader -w mysqlwriter
"job":
"content": [
"reader":
"name": "mongodbreader", # 读取端
"parameter":
"address": [], # mongodb 地址
"collectionName":"", # 集合名称
"column": [], # 字段名称
"dbName": "", # 数据库名称
"userPassword": "" # 密码
,
"writer":
"name": "mysqlwriter", # 写入端
"parameter":
"column": [], # 需要同步的列
"connection": [
"jdbcUrl": "", # 连接信息
"table": [] # 连接表
],
"password": "", # 连接密码
"preSql": [], # 同步前. 要做的事
"session": [],
"username": "", # 连接用户
"writeMode": "" # 操作类型
],
"setting":
"speed":
"channel": "" # 指定并发数
第二步：编写Mongodb数据同步文件:MongodbToMySQL.json
"job":
"content": [
"reader":
"name": "mongodbreader",
"parameter":
"address": ["*************:27017"],
"collectionName": "mongo2mysql",
"column": [
"name":"name",
"type":"string"
,
"name":"age",
"type":"double"
,
"name":"gender",
"type":"string"
],
"dbName": "test"
,
"writer":
"name": "mysqlwriter",
"parameter":
"column": ["*"],
"connection": [
"jdbcUrl": "jdbc:mysql://*************:3306/datax",
"table": ["test"]
],
"password": "*******",
"username": "root",
"writeMode": "insert"
],
"setting":
"speed":
"channel": "1"
第三步：验证
python /usr/local/datax/bin/datax.py ./MongodbToMySQL.json 控制台输出：
3.4 通过 DataX 实现 Oracle数据同步MySQL
第一步：生成 Oracle 到 MySQL数据同步的模板。
[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py -r oraclereader -w mysqlwriter
"job":
"content": [
"reader":
"name": "oraclereader", # 读取端
"parameter":
"column": [], # 字段名称
"connection":[
"jdbcUrl": [], # 数据库url
"table": [] # 表名
],
"password": "", # oracle 密码
"username": "", # oracle 账户
,
"writer":
"name": "mysqlwriter", # 写入端
"parameter":
"column": [], # 需要同步的列
"connection": [
"jdbcUrl": "", # 连接信息
"table": [] # 连接表
],
"password": "", # 连接密码
"preSql": [], # 同步前. 要做的事
"session": [],
"username": "", # 连接用户
"writeMode": "" # 操作类型
],
"setting":
"speed":
"channel": "" # 指定并发数
第二步：编写Oracle数据同步文件:OracleToMySQL.json (全量同步)
"job":
"content": [
"reader":
"name": "oraclereader",
"parameter":
"column": ["INVESTOR_ID","INVESTOR_NAME","ID_TYPE","ID_NO","CREATE_TIME"],
"splitPk": "INVESTOR_ID",
"where" : "INVESTOR_ID is not null",
"connection": [
"jdbcUrl": ["jdbc:oracle:thin:@172.17.112.177:1521:helowin"],
"table": ["CXX.CUSTOMER"]
],
"password": "123456",
"username": "admin"
,
"writer":
"name": "mysqlwriter",
"parameter":
"column": [
"customer_no",
"customer_name",
"id_type",
"id_no",
"create_time"
],
"connection": [
"jdbcUrl": "jdbc:mysql://172.17.112.176:3306/customer_db?useUnicode=true&characterEncoding=UTF-8",
"table": ["customer_datax"]
],
"username": "admin",
"password": "123456",
"preSql": ["truncate table customer_datax"],
"writeMode": "insert"
],
"setting":
"speed":
"channel": "3"
第三步：控制台执行全量同步脚本
python /usr/local/datax/bin/datax.py ./OracleToMySQL.json 四、Datax 开发实战
实际功能需求：产品经理要求实现，除第一次Datax 全量同步Oracle 至MySQL数据外，还要求每天自动同步Oracle 数据至MySQL数据库。
功能设计关键点
- datax 增量同步脚本
- Shell 脚本调用Python 脚本，并动态传入相关 参数
- Linux 系统定时任务Cron 自动调用Shell 脚本 或者Java 分布式定时任务框架XXL-JOB 调用Shelll脚本
第一步：编写Oracle 同步MySQL 同步全量脚本, Trans_All_House.json
# 创建Trans_All_House.json
"job":
"content": [
"reader":
"name": "oraclereader",
"parameter":
"splitPK": "LSBH",
"connection": [
"jdbcUrl": ["jdbc:oracle:thin:@172.17.112.177:1521:helowin"],
"querySql": ["select LSHB, YWBH, TJSBYWBH, CHYWLB, FWBM, ZRZBH, BDCDYDM, FWZT, SZQSC, SZZZC, MYC, DY, FH, FWZL, ****, to_char(YWBJSJ, 'yyyy-MM-dd HH24:mm:ss') as YWBJSJ, to_char(SJCRSJ, 'yyyy-MM-dd HH24:mm:ss') as SJCRSJ from BASE_HOUSE"]
],
"password": "123456",
"username": "admin"
,
"writer":
"name": "mysqlwriter",
"parameter":
"column": [
"LSHB",
"YWBH",
"TJSBYWBH",
"CHYWLB",
"FWBM",
"ZRZBH",
"BDCDYDM",
"FWZT",
"SZQSC",
"SZZZC",
"MYC",
"DY",
"FH",
"FWZL",
******
"YWBJSJ",
"SJCRSJ"
],
"connection": [
"jdbcUrl": "jdbc:mysql://172.17.112.176:3306/customer_db?useUnicode=true&characterEncoding=UTF-8",
"table": ["BASE_HOUSE"]
],
"username": "admin",
"password": "123456",
"preSql": ["truncate table BASE_HOUSE"],
"writeMode": "insert"
],
"setting":
"speed":
"channel": "3"
第二步：编写Shell 脚本调用全量同步脚本
# 编辑全量同步Shell脚本
vi trans_all_house.sh
# 文本内容
#!/bin/bash
source /etc/profile
# 执行datax脚本，传入时间范围
python /usr/local/datax/bin/datax.py usr/local/datax/job/Trans_All_House.json
温馨提示：
1、为trans_all_house.sh 脚本添加可执行权限
chmod +777 trans_all_house.sh2、执行trans_all_house.sh 脚本
./trans_all_house.sh
第三步：编写Oracle 同步MySQL 同步增量脚本, Increase_Day_House.json
# 创建Increase_Day_House.json
"job":
"content": [
"reader":
"name": "oraclereader",
"parameter":
"splitPK": "LSBH",
"connection": [
"jdbcUrl": ["jdbc:oracle:thin:@172.17.112.177:1521:helowin"],
"querySql": ["select LSHB, YWBH, TJSBYWBH, CHYWLB, FWBM, ZRZBH, BDCDYDM, FWZT, SZQSC, SZZZC, MYC, DY, FH, FWZL, ****, to_char(YWBJSJ, 'yyyy-MM-dd HH24:mm:ss') as YWBJSJ, to_char(SJCRSJ, 'yyyy-MM-dd HH24:mm:ss') as SJCRSJ from BASE_HOUSE where CREATE_DT >='$start_time 00:00:00' and CREATE_DT <='$end_time 23:59:59'"]
],
"password": "123456",
"username": "admin"
,
"writer":
"name": "mysqlwriter",
"parameter":
"column": [
"LSHB",
"YWBH",
"TJSBYWBH",
"CHYWLB",
"FWBM",
"ZRZBH",
"BDCDYDM",
"FWZT",
"SZQSC",
"SZZZC",
"MYC",
"DY",
"FH",
"FWZL",
******
"YWBJSJ",
"SJCRSJ"
],
"connection": [
"jdbcUrl": "jdbc:mysql://172.17.112.176:3306/customer_db?useUnicode=true&characterEncoding=UTF-8",
"table": ["BASE_HOUSE"]
],
"username": "admin",
"password": "123456",
"writeMode": "update"
],
"setting":
"speed":
"channel": "3"
第四步：编写Shell 脚本调用增量同步脚本
# 编辑增量同步Shell脚本
vi increase_day_house.sh
增量同步Shell 脚本内容：
#!/bin/bash
source /etc/profile
# 截至时间设置为当前时间戳
end_time=$(date +%s)
# 开始时间设置为300s前时间戳
create_time=$(($end_time - 300))
# 执行datax脚本，传入时间范围
python /usr/local/datax/bin/datax.py /usr/local/datax/job/increase_day_house.json -p "-Dcreate_time=$create_time -Dend_time=$end_time" &
第五步：设置 crontab 定时任务，每天23点执行一次。
#修改 crontab 文件，如果文件不存在会自动创建。
crontab -e
# 添加Shell 脚本定时任务调度
0 0 23 * * ? /usr/local/datax/sh/increment_day_house.sh >/dev/null 2>&1以上是关于DataX框架的设计、运行原理详解的主要内容，如果未能解决你的问题，请参考以下文章