【大数据】DataX 部署及使用
DataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能👉点击前往查看源码✍简单了解了DataX的使用方法，全量表数据由DataX从MySQL业务数据库直接同步到HDFS。
前言
DataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能
一、DataX简介
1. DataX架构原理
🚄DataX设计理念：为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步
🚄DataX框架设计：DataX本身作为离线数据同步框架，采用Framework + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中
2. DataX运行流程
🚄DataX运行流程：下面用一个DataX作业生命周期的时序图说明DataX的运行流程、核心概念以及每个概念之间的关系
🚄DataX调度决策思路：举例来说，用户提交了一个DataX作业，并且配置了总的并发度为20，目的是对一个有100张分表的mysql数据源进行同步。DataX的调度决策思路是：
- DataX Job根据分库分表切分策略，将同步工作分成100个Task
- 根据配置的总的并发度20，以及每个Task Group的并发度5，DataX计算共需要分配4个TaskGroup
- 4个TaskGroup平分100个Task，每一个TaskGroup负责运行25个Task
🚄DataX与Sqoop对比
| 功能 | DataX | Sqoop |
|---|---|---|
| 运行模式 | 单进程多线程 | MR |
| 分布式 | 不支持，可以通过调度系统规避 | 支持 |
| 流控 | 有流控功能 | 需要定制 |
| 统计信息 | 已有一些统计，上报需定制 | 没有，分布式的数据收集不方便 |
| 数据校验 | 在core部分有校验功能 | 没有，分布式的数据收集不方便 |
| 监控 | 需要定制 | 需要定制 |
二、DataX部署及使用
1. 解压即可使用
# 下载
cd /opt/software;wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
# 解压
tar -zxvf datax.tar.gz -C /opt/module/
cd /opt/module/;ll
# 自检，执行如下命令
python /opt/module/datax/bin/datax.py /opt/module/datax/job/job.json
🤯遗憾报错：
😅解决方案：
# /opt/module目录下执行即可
rm -rf datax/plugin/*/._*
2. DataX任务提交命令
🚄DataX的使用十分简单，用户只需根据自己同步数据的数据源和目的地选择相应的Reader和Writer，并将Reader和Writer的信息配置在一个json文件中，然后执行如下命令提交数据同步任务即可
# 格式，只需提供一个json文件
python bin/datax.py path/to/your/job.json
3. DataX配置文件格式
# 查看DataX配置文件模板
cd /opt/module/datax
python bin/datax.py -r mysqlreader -w hdfswriter
👇配置文件模板如下，json最外层是一个job，job包含setting和content两部分，其中setting用于对整个job进行配置，content用户配置数据源和目的地：
🚄DataX支持的数据源：DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入，目前支持数据如下，👉点击前往官方文档
| 类型 | 数据源 | 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 | √ | √ |
4. 同步MySQL数据➡HDFS案例
- 案例要求：同步gmall数据库中base_province表数据到HDFS的/base_province目录
- 需求分析：要实现该功能，需选用MySQLReader和HDFSWriter，MySQLReader具有两种模式分别是
TableMode和QuerySQLMode，前者使用table，column，where等属性声明需要同步的数据；后者使用一条SQL查询语句声明需要同步的数据。下面分别使用两种模式进行演示
4.1. MySQLReader之TableMode
🚄创建配置文件vim /opt/module/datax/job/base_province.json（记得修改一下MySQL密码）
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"column": [
"id",
"name",
"region_id",
"area_code",
"iso_code",
"iso_3166_2"
],
"where": "id>=3",
"connection": [
{
"jdbcUrl": [
"jdbc:mysql://hadoop102:3306/gmall"
],
"table": [
"base_province"
]
}
],
"password": "修改成自己MySQL的密码",
"splitPk": "",
"username": "root"
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"column": [
{
"name": "id",
"type": "bigint"
},
{
"name": "name",
"type": "string"
},
{
"name": "region_id",
"type": "string"
},
{
"name": "area_code",
"type": "string"
},
{
"name": "iso_code",
"type": "string"
},
{
"name": "iso_3166_2",
"type": "string"
}
],
"compress": "gzip",
"defaultFS": "hdfs://hadoop102:8020",
"fieldDelimiter": "\t",
"fileName": "base_province",
"fileType": "text",
"path": "/base_province",
"writeMode": "append"
}
}
}
],
"setting": {
"speed": {
"channel": 1
}
}
}
}
🚄配置文件说明
- Reader参数说明
- Writer参数说明
🤯注意事项：
HFDS Writer并未提供nullFormat参数：也就是用户并不能自定义null值写到HFDS文件中的存储格式- 默认情况下，
HFDS Writer会将null值存储为空字符串（‘’），而Hive默认的null值存储格式为\N。所以后期将DataX同步的文件导入Hive表就会出现问题。
😅解决方案(2种)：
- 修改DataX HDFS Writer的源码，增加自定义null值存储格式的逻辑，可参考这篇blog
- 在Hive中建表时指定null值存储格式为空字符串（‘’），例如：
DROP TABLE IF EXISTS base_province;
CREATE EXTERNAL TABLE base_province
(
`id` STRING COMMENT '编号',
`name` STRING COMMENT '省份名称',
`region_id` STRING COMMENT '地区ID',
`area_code` STRING COMMENT '地区编码',
`iso_code` STRING COMMENT '旧版ISO-3166-2编码，供可视化使用',
`iso_3166_2` STRING COMMENT '新版IOS-3166-2编码，供可视化使用'
) COMMENT '省份表'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
NULL DEFINED AS ''
LOCATION '/base_province/';
- Setting参数说明
🚄测试一下：
# 要在HDFS创建/base_province目录
hadoop fs -mkdir /base_province
# 启动
cd /opt/module/datax;python bin/datax.py job/base_province.json
🤯发现报错：
😅原来是缺少MySQL8的驱动jar包，8之前的不会有这个问题，之前下了👉点击前往下载，最后放在datax/lib/目录下即可
# 查看HDFS文件
hadoop fs -cat /base_province/* | zcat
4.2. MySQLReader之QuerySQLMode
🔨修改配置文件base_province.json，直接删了贴如下内容过去（记得修改一下MySQL密码）
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"connection": [
{
"jdbcUrl": [
"jdbc:mysql://hadoop102:3306/gmall"
],
"querySql": [
"select id,name,region_id,area_code,iso_code,iso_3166_2 from base_province where id>=3"
]
}
],
"password": "000000",
"username": "root"
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"column": [
{
"name": "id",
"type": "bigint"
},
{
"name": "name",
"type": "string"
},
{
"name": "region_id",
"type": "string"
},
{
"name": "area_code",
"type": "string"
},
{
"name": "iso_code",
"type": "string"
},
{
"name": "iso_3166_2",
"type": "string"
}
],
"compress": "gzip",
"defaultFS": "hdfs://hadoop102:8020",
"fieldDelimiter": "\t",
"fileName": "base_province",
"fileType": "text",
"path": "/base_province",
"writeMode": "append"
}
}
}
],
"setting": {
"speed": {
"channel": 1
}
}
}
}
🚄配置文件说明
- Reader参数说明
- Writer、Setting参数一致，不同只有Reader部分，一个用结构，一个用select语句（无法用splitPk，但更灵活）
🚄测试一下：
# 启动
cd /opt/module/datax;python bin/datax.py job/base_province.json
# 查看HDFS
hadoop fs -cat /base_province/base_province__71f91e1e_0d61_4b81_9570_f355badd5ada.gz | zcat
4.3. DataX传参
通常情况下，离线数据同步任务需要每日定时重复执行，故HDFS上的目标路径通常会包含一层日期，以对每日同步的数据加以区分，也就是说每日同步数据的目标路径不是固定不变的，因此DataX配置文件中HDFS Writer的path参数的值应该是动态的。为实现这一效果，就需要使用DataX传参的功能
🔨DataX传参的用法如下，在JSON配置文件中使用${param}引用参数，在提交任务时使用-p"-Dparam=value"传入参数值，具体示例如下，修改配置文件base_province.json（记得修改一下MySQL密码）
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"connection": [
{
"jdbcUrl": [
"jdbc:mysql://hadoop102:3306/gmall"
],
"querySql": [
"select id,name,region_id,area_code,iso_code,iso_3166_2 from base_province where id>=3"
]
}
],
"password": "000000",
"username": "root"
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"column": [
{
"name": "id",
"type": "bigint"
},
{
"name": "name",
"type": "string"
},
{
"name": "region_id",
"type": "string"
},
{
"name": "area_code",
"type": "string"
},
{
"name": "iso_code",
"type": "string"
},
{
"name": "iso_3166_2",
"type": "string"
}
],
"compress": "gzip",
"defaultFS": "hdfs://hadoop102:8020",
"fieldDelimiter": "\t",
"fileName": "base_province",
"fileType": "text",
"path": "/base_province/${dt}",
"writeMode": "append"
}
}
}
],
"setting": {
"speed": {
"channel": 1
}
}
}
}
🚄测试一下：
# 创建目标路径
hadoop fs -mkdir /base_province/2020-06-14
# 进入DataX根目录
cd /opt/module/datax;python bin/datax.py -p"-Ddt=2020-06-14" job/base_province.json
# 查看结果
hadoop fs -ls /base_province
5. 同步HDFS数据➡MySQL案例
- 案例要求：同步HDFS上的
/base_province目录下的数据到MySQL gmall数据库下的test_province表 - 需求分析：要实现该功能，需选用
HDFSReader和MySQLWriter
🔨修改配置文件，vim /opt/module/datax/job/base_province.json
{
"job": {
"content": [
{
"reader": {
"name": "hdfsreader",
"parameter": {
"defaultFS": "hdfs://hadoop102:8020",
"path": "/base_province",
"column": [
"*"
],
"fileType": "text",
"compress": "gzip",
"encoding": "UTF-8",
"nullFormat": "\\N",
"fieldDelimiter": "\t",
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"username": "root",
"password": "000000",
"connection": [
{
"table": [
"test_province"
],
"jdbcUrl": "jdbc:mysql://hadoop102:3306/gmall?useUnicode=true&characterEncoding=utf-8"
}
],
"column": [
"id",
"name",
"region_id",
"area_code",
"iso_code",
"iso_3166_2"
],
"writeMode": "replace"
}
}
}
],
"setting": {
"speed": {
"channel": 1
}
}
}
}
🚄配置文件说明
- Reader参数说明
- Writer参数说明
🚄测试一下：
-- 在MySQL gmall中创建gmall.test_province表
DROP TABLE IF EXISTS `test_province`;
CREATE TABLE `test_province` (
`id` bigint(20) NOT NULL,
`name` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`region_id` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`area_code` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`iso_code` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`iso_3166_2` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;
# 进入DataX根目录
cd /opt/module/datax;python bin/datax.py job/base_province.json
👇进行了去重后的数据
三、DataX优化
1. 速度控制
🚄DataX3.0提供了包括通道(并发)、记录流、字节流三种流控模式，可以随意控制你的作业速度，让你的作业在数据库可以承受的范围内达到最佳的同步速度。关键优化参数如下：
| 参数 | 说明 |
|---|---|
| job.setting.speed.channel | 并发数 |
| job.setting.speed.record | 总record限速 |
| job.setting.speed.byte | 总byte限速 |
| core.transport.channel.speed.record | 单个channel的record限速，默认值为10000（10000条/s） |
| core.transport.channel.speed.byte | 单个channel的byte限速，默认值1024*1024（1M/s） |
注意事项：
- 若配置了总record限速，则必须配置单个channel的record限速
- 若配置了总byte限速，则必须配置单个channe的byte限速
- 若配置了总record限速和总byte限速，channel并发数参数就会失效。因为配置了总record限速和总byte限速之后，实际channel并发数是通过计算得到的：
- 计算公式为：min(总byte限速/单个channel的byte限速，总record限速/单个channel的record限速)
配置示例：
{
"core": {
"transport": {
"channel": {
"speed": {
"byte": 1048576 //单个channel byte限速1M/s
}
}
}
},
"job": {
"setting": {
"speed": {
"byte" : 5242880 //总byte限速5M/s
}
},
...
}
}
2. 内存调整
当提升DataX Job内Channel并发数时，内存的占用会显著增加，因为DataX作为数据交换通道，在内存中会缓存较多的数据。例如Channel中会有一个Buffer，作为临时的数据交换的缓冲区，而在部分Reader和Writer的中，也会存在一些Buffer，为了防止OOM等错误，需调大JVM的堆内存
🚄建议将内存设置为4G或者8G，这个也可以根据实际情况来调整。调整JVM xms xmx参数的两种方式
- 一种是直接更改datax.py脚本
- 另一种是在启动的时候，加上对应的参数，如下：
python datax/bin/datax.py --jvm="-Xms8G -Xmx8G" /path/to/your/job.json
总结
✍简单了解了DataX的使用方法，全量表数据由DataX从MySQL业务数据库直接同步到HDFS
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
0
0- 0
已为社区贡献1条内容
所有评论(0)