大数据-数据源离线同步工具DataX的使用
前言
官方网址https://github.com/alibaba/DataX
DataX 是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。
为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。
设计理念
DataX本身作为离线数据同步框架，采用Framework + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。
- Reader：Reader为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
- Writer： Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
- Framework：Framework用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。
支持的数据类型
类型 | 数据源 | Reader(读) | Writer(写) | 文档 |
|---|---|---|---|---|
RDBMS 关系型数据库 | MySQL | √ | √ | 读 、写 |
Oracle | √ | √ | 读 、写 | |
OceanBase | √ | √ | 读 、写 | |
SQLServer | √ | √ | 读 、写 | |
PostgreSQL | √ | √ | 读 、写 | |
DRDS | √ | √ | 读 、写 | |
通用RDBMS(支持所有关系型数据库) | √ | √ | 读 、写 | |
阿里云数仓数据存储 | ODPS | √ | √ | 读 、写 |
ADS | √ | 写 | ||
OSS | √ | √ | 读 、写 | |
OCS | √ | 写 | ||
NoSQL数据存储 | OTS | √ | √ | 读 、写 |
Hbase0.94 | √ | √ | 读 、写 | |
Hbase1.1 | √ | √ | 读 、写 | |
Phoenix4.x | √ | √ | 读 、写 | |
Phoenix5.x | √ | √ | 读 、写 | |
MongoDB | √ | √ | 读 、写 | |
Hive | √ | √ | 读 、写 | |
Cassandra | √ | √ | 读 、写 | |
无结构化数据存储 | TxtFile | √ | √ | 读 、写 |
FTP | √ | √ | 读 、写 | |
HDFS | √ | √ | 读 、写 | |
Elasticsearch | √ | 写 | ||
时间序列数据库 | OpenTSDB | √ | 读 | |
TSDB | √ | √ | 读 、写 | |
TDengine | √ | √ | 读 、写 |
DataX3.0核心架构
DataX 3.0 开源版本支持单机多线程模式完成同步作业运行，本小节按一个DataX作业生命周期的时序图，从整体架构设计非常简要说明DataX各个模块相互关系。
举例来说，用户提交了一个DataX作业，并且配置了20个并发，目的是将一个100张分表的mysql数据同步到odps里面。
DataX的调度决策思路是：
- DataXJob根据分库分表切分成了100个Task。
- 根据20个并发，DataX计算共需要分配4个TaskGroup。
- 4个TaskGroup平分切分好的100个Task，每一个TaskGroup负责以5个并发共计运行25个Task。
工具部署
环境
- Linux
- JDK(1.8以上，推荐1.8)
- Python(2或3都可以)
- DataX 编译后版本
直接下载编译后的DataX工具包：DataX下载地址
下载后解压至本地某个目录，进入bin目录，即可运行同步作业：
解压到目录
/data/tools/bigdata/datax3
配置环境变量
创建配置文件
vi /etc/profile.d/datax.sh内容设置为
# DataX
export DATAX_HOME=/data/tools/bigdata/datax3
export PATH=$PATH:$DATAX_HOME/bin配置生效
source /etc/profile查看是否生效
echo $DATAX_HOME运行方式
cd $DATAX_HOME/bin/
python datax.py YOUR_JOB.json自检脚本：
python $DATAX_HOME/bin/datax.py $DATAX_HOME/job/job.json定时任务
crontab 命令
-l在标准输出上显示当前的crontab。-r删除当前的crontab文件。-e使用VISUAL或者EDITOR环境变量所指的编辑器编辑当前的crontab文件。当结束编辑离开时，编辑后的文件将自动安装。
格式如下：
minute hour day-of-month month-of-year day-of-week commands合法值
00-59 00-23 01-31 01-12 0-6 (0 is sunday)除了数字还有几个个特殊的符号就是*、/和-、,，*代表所有的取值范围内的数字
/代表每的意思,/5表示每5个单位-代表从某个数字到某个数字,分开几个离散的数字
查看定时任务
crontab -l添加定时任务
crontab -e添加如下
5 1 * * * python $DATAX_HOME/bin/datax.py $DATAX_HOME/config/oracle2oracle.json >>$DATAX_HOME/log/datax_log.`date +\%Y\%m\%d\%H\%M\%S` 2>&1或者自己添加配置文件
crontab /data/cron/mysqlRollBack.cron配置
配置项
- job 下面有两个配置项，content 和 setting，其中 content 用来描述该任务的源和目的端的信息，setting 用来描述任务本身的信息；
- content 又分为两部分，reader 和 writer，分别用来描述源端和目的端的信息；
- setting 中的 speed 项表示同时起几个并发执行该任务；
job.setting.speed（流量控制） Job 支持用户对速度的自定义控制，channel 的值可以控制同步时的并发数，byte 的值可以控制同步时的速度。job.setting.errorLimit（脏数据控制） Job 支持用户对于脏数据的自定义监控和告警，包括对脏数据最大记录数阈值（record 值）或者脏数据占比阈值（percentage 值），当 Job 传输过程出现的脏数据大于用户指定的数量/百分比，DataX Job 报错退出。
示例
{
"job": {
"setting": {
"speed": {
"channel": 3
},
"errorLimit": {
"record": 0,
"percentage": 0.02
}
},
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"username": "root",
"password": "root",
"column": [
"id",
"name"
],
"splitPk": "db_id",
"connection": [
{
"table": [
"table"
],
"jdbcUrl": [
"jdbc:mysql://127.0.0.1:3306/database"
]
}
]
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"print":true
}
}
}
]
}
}测试
Stream ==> Stream
使用 streamreader + streamwriter（这种情况常用于测试）
配置文件：stream2stream.json
{
"job": {
"content": [{
"reader": {
"name": "streamreader",
"parameter": {
"sliceRecordCount": 10,
"column": [{
"type": "String",
"value": "hello DataX"
}, {
"type": "string",
"value": "DataX Stream To Stream"
}, {
"type": "string",
"value": "数据迁移工具"
}]
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"encoding": "GBK",
"print": true
}
}
}],
"setting": {
"speed": {
"channel": 2
}
}
}
}输入执行命令：
python $DATAX_HOME/bin/datax.py ../job/stream2stream.json