一、介绍
DataX 是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。
github地址
详细文档
操作手册
支持数据框架如下：
架构
Reader：为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
Writer：为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
Framework：用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。
二、使用
- 下载
下载地址：https://datax-opensource.oss-cn-hangzhou.aliyuncs.com/202308/datax.tar.gz
- 解压缩
# 解压缩
tar -zxvf datax.tar.gz -C /opt/module/
- 编写数据同步任务
{
"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
}
}
}
}
- 启动任务
python /opt/module/datax/bin/datax.py /opt/module/datax/job/stream_to_stream.json
- 执行结果
- 配置说明
| 参数 | 说明 |
|---|---|
| job.setting | 设置全局配置参数 |
| job.setting.speed | 控制任务速度配置参数,包括：channel(通道(并发))、record(字节流)、byte(记录流)等三种模式 |
| job.setting.speed.channel | 并发数 |
| job.setting.speed.record | 字节流 |
| job.setting.speed.byte | 记录流 |
| job.setting.errorLimit | 设置错误限制 |
| job.setting.errorLimit.record | 指定允许的最大错误记录数 |
| job.setting.errorLimit.percentage | 指定允许的最大错误记录百分比 |
| job.setting.dirtyDataPath | 设置错误限制 |
| job.setting.dirtyDataPath.path | 设置错误限制 |
| job.setting.log | 设置错误限制 |
| job.setting.log.level | 设置错误限制 |
| job.setting.log.dir | 设置错误限制 |
| content | 任务配置参数 |
| reader | Reader配置 |
| name | Reader类型 |
| parameter | Reader具体配置(具体配置查看具体Reader) |
| writer | Writer配置 |
| name | Writer类型 |
| parameter | Writer具体配置(具体配置查看具体Writer) |
三、常用配置
3.1、MysqlReader
{
"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
}
}
}
]
}
}
配置说明：
jdbcUrl：链接地址
username：mysql用户名
password：mysql密码
table：待同步的表名
column：所配置的表中需要同步的列名集合，可以使用使用*代表所有字段
splitPk：使用splitPk代表的字段进行数据分片，DataX因此会启动并发任务进行数据同步，这样可以大大提供数据同步的效能
where：筛选条件
querySql：sql语句，可以替代column和where配置
3.2、MysqlWriter
{
"job": {
"setting": {
"speed": {
"channel": 1
}
},
"content": [
{
"reader": {
"name": "streamreader",
"parameter": {
"column" : [
{
"value": "DataX",
"type": "string"
},
{
"value": 19880808,
"type": "long"
},
{
"value": "1988-08-08 08:08:08",
"type": "date"
},
{
"value": true,
"type": "bool"
},
{
"value": "test",
"type": "bytes"
}
],
"sliceRecordCount": 1000
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"writeMode": "insert",
"username": "root",
"password": "root",
"column": [
"id",
"name"
],
"session": [
"set session sql_mode='ANSI'"
],
"preSql": [
"delete from test"
],
"connection": [
{
"jdbcUrl": "jdbc:mysql://127.0.0.1:3306/datax?useUnicode=true&characterEncoding=gbk",
"table": [
"test"
]
}
]
}
}
}
]
}
}
配置说明：
jdbcUrl：链接地址
username：mysql用户名
password：mysql密码
table：待同步的表名
column：所配置的表中需要同步的列名集合，可以使用使用*代表所有字段
preSql：写入数据到目的表前，会先执行这里的标准语句
postSql：写入数据到目的表后，会执行这里的标准语句
writeMode：控制写入数据到目标表采用insert into或者replace into或者 ON DUPLICATE KEY UPDATE语句
batchSize：一次性批量提交的记录数大小
3.3、HdfsReader
{
"job": {
"setting": {
"speed": {
"channel": 3
}
},
"content": [
{
"reader": {
"name": "hdfsreader",
"parameter": {
"path": "/user/hive/warehouse/mytable01/*",
"defaultFS": "hdfs://xxx:port",
"column": [
{
"index": 0,
"type": "long"
},
{
"index": 1,
"type": "boolean"
},
{
"type": "string",
"value": "hello"
},
{
"index": 2,
"type": "double"
}
],
"fileType": "orc",
"encoding": "UTF-8",
"fieldDelimiter": ","
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}
]
}
}
配置说明：
path：文件路径
defaultFS：namenode节点地址
fileType：文件的类型，目前支持：”text”、”orc”、”rc”、”seq”、”csv”
column：读取字段列表
fieldDelimiter：读取的字段分隔符
encoding：读取文件的编码配置
nullFormat：文本文件中无法使用标准字符串定义null(空指针)，DataX提供nullFormat定义哪些字符串可以表示为null
haveKerberos：是否有Kerberos认证，默认false
kerberosKeytabFilePath：Kerberos认证keytab文件路径，且为绝对路径
kerberosPrincipal：Kerberos认证Principal名
hadoopConfig：hadoop相关的一些高级参数
3.4、HdfsWriter
{
"setting": {},
"job": {
"setting": {
"speed": {
"channel": 2
}
},
"content": [
{
"reader": {
"name": "txtfilereader",
"parameter": {
"path": ["/Users/shf/workplace/txtWorkplace/job/dataorcfull.txt"],
"encoding": "UTF-8",
"column": [
{
"index": 0,
"type": "long"
},
{
"index": 1,
"type": "long"
},
{
"index": 2,
"type": "long"
},
{
"index": 3,
"type": "long"
},
{
"index": 4,
"type": "DOUBLE"
},
{
"index": 5,
"type": "DOUBLE"
},
{
"index": 6,
"type": "STRING"
},
{
"index": 7,
"type": "STRING"
},
{
"index": 8,
"type": "STRING"
},
{
"index": 9,
"type": "BOOLEAN"
},
{
"index": 10,
"type": "date"
},
{
"index": 11,
"type": "date"
}
],
"fieldDelimiter": "\t"
}
},
"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"
},
{
"name": "col4",
"type": "BIGINT"
},
{
"name": "col5",
"type": "FLOAT"
},
{
"name": "col6",
"type": "DOUBLE"
},
{
"name": "col7",
"type": "STRING"
},
{
"name": "col8",
"type": "VARCHAR"
},
{
"name": "col9",
"type": "CHAR"
},
{
"name": "col10",
"type": "BOOLEAN"
},
{
"name": "col11",
"type": "date"
},
{
"name": "col12",
"type": "TIMESTAMP"
}
],
"writeMode": "append",
"fieldDelimiter": "\t",
"compress":"NONE"
}
}
}
]
}
}
配置说明：
path：存储到Hadoop Hdfs文件系统的路径信息
defaultFS：namenode节点地址
fileType：文件的类型，目前支持：“text”或“orc”
fileName：文件名
column：写入字段列表
fieldDelimiter：读取的字段分隔符
compress：文件压缩类型
3.5、FtpReader
{
"setting": {},
"job": {
"setting": {
"speed": {
"channel": 2
}
},
"content": [
{
"reader": {
"name": "ftpreader",
"parameter": {
"protocol": "sftp",
"host": "127.0.0.1",
"port": 22,
"username": "xx",
"password": "xxx",
"path": [
"/home/hanfa.shf/ftpReaderTest/data"
],
"column": [
{
"index": 0,
"type": "long"
},
{
"index": 1,
"type": "boolean"
},
{
"index": 2,
"type": "double"
},
{
"index": 3,
"type": "string"
},
{
"index": 4,
"type": "date",
"format": "yyyy.MM.dd"
}
],
"encoding": "UTF-8",
"fieldDelimiter": ","
}
},
"writer": {
"name": "ftpWriter",
"parameter": {
"path": "/home/hanfa.shf/ftpReaderTest/result",
"fileName": "shihf",
"writeMode": "truncate",
"format": "yyyy-MM-dd"
}
}
}
]
}
}
配置说明：
protocol：ftp服务器协议，目前支持传输协议有ftp和sftp
host：ftp服务器地址
port：ftp服务器端口
timeout：连接ftp服务器连接超时时间，单位毫秒，默认：60000
connectPattern：连接模式（主动模式或者被动模式）
username：用户名
password：密码
path：路径
column：读取字段列表
fieldDelimiter：读取的字段分隔符
3.6、FtpWriter
{
"setting": {},
"job": {
"setting": {
"speed": {
"channel": 2
}
},
"content": [
{
"reader": {},
"writer": {
"name": "ftpwriter",
"parameter": {
"protocol": "sftp",
"host": "***",
"port": 22,
"username": "xxx",
"password": "xxx",
"timeout": "60000",
"connectPattern": "PASV",
"path": "/tmp/data/",
"fileName": "yixiao",
"writeMode": "truncate|append|nonConflict",
"fieldDelimiter": ",",
"encoding": "UTF-8",
"nullFormat": "null",
"dateFormat": "yyyy-MM-dd",
"fileFormat": "csv",
"suffix": ".csv",
"header": []
}
}
}
]
}
}
配置说明：
protocol：ftp服务器协议，目前支持传输协议有ftp和sftp
host：ftp服务器地址
port：ftp服务器端口
timeout：连接ftp服务器连接超时时间，单位毫秒，默认：60000
connectPattern：连接模式（主动模式或者被动模式）
username：用户名
password：密码
path：路径
fileName:文件名
fieldDelimiter：读取的字段分隔符
