DataX入门———用法及一些简单的案例介绍
- 一、概述
-
- 1. 简介
- 2. 设计理念
- 3. 运行框架
- 二、安装
- 三、案例介绍
-
- 1. 从stream流读取数据并打印到控制台
- 2. 读取mysql数据存放到hdfs
- 3. 读取HDFS数据存放到MySQL
- 4. 读取Oralce数据存放到Mysql
一、概述
1. 简介
DataX是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库（MySQL、Oracle）、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。
优点，使用方便，效率高，并且打印同步时间，同步数据的日志，并且都是中文，方便查看。
2. 设计理念
为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要介入一个新的数据源的时候，只需要将此数据对接到DataX，便能跟已有的数据源做到无缝数据同步。
3. 运行框架
Reader：数据采集模块，负责采集数据源的数据，将数据发送给Framework。
Writer：数据写入模块，负责不断向Framework取数据，并将数据写入到目的地端。
Framework：用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。
二、安装
安装包百度云下载：https://pan.baidu.com/s/1oi3qFX-bfvyTt4V-1AJ7bw
提取码：
3z3w
前置条件
JDK1.8 以上
python2.6 以上
解压到指定目录即可
[[email protected] ~]# tar -zxvf /opt/datax.tar.gz -C /opt/soft/
检测能否正常运行
[[email protected] ~]# cd /opt/soft/datax/
[[email protected] datax]# bin/datax.py job/job.json
如下所示为显示正常
下面做一些简单案例介绍，
操作案例也可以访问官方地址，极其详细：https://github.com/alibaba/DataX
三、案例介绍
1. 从stream流读取数据并打印到控制台
输入如下指令可以查看官方配置模板
[[email protected] datax]# bin/datax.py -r streamreader -w streamwriter
例：
sliceRecordCount，表示切片为10，即打印10次
column，打印哪种类型type，值是多少value
setting配置下channel设置为1，表示并发量为1
{
"job": {
"content": [
{
"reader": {
"name": "streamreader",
"parameter": {
"column": [
{
"type":"long",
"value":"10"
},
{
"type":"string",
"value":"hello,我的第一个DataX"
}
],
"sliceRecordCount": 10
}
},
"writer": {
"name": "streamwriter",
"parameter": {
"encoding": "UTF-8",
"print": true
}
}
}
],
"setting": {
"speed": {
"channel":"1"
}
}
}
}
创建以上案例配合为json文件
# 复制以上内容到json文件
vi job/stream2stream.json
执行
[[email protected] datax]# bin/datax.py job/stream2stream.json
如下控制台显示打印效果以及读写时间记录
2. 读取mysql数据存放到hdfs
输入如下指令可以查看官方配置模板
[[email protected] datax]# bin/datax.py -r mysqlreader -w hdfswriter
读取mysql数据库mytest下的temperatures表
启动hdfs
[[email protected] datax]# start-all.sh
创建用于上传的hdfs目录
[[email protected] datax]# hdfs dfs -mkdir /mydataxInfo
mysqlreader部分配置参数说明
column：需要同步的列名集合，使用JSON数组描述自带信息，*代表所有列
jdbcUrl：对数据库的JDBC连接信息，使用JSON数组描述，支持多个连接地址
table：需要同步的表，支持多个
querySql：自定义SQL，配置该属性会直接忽略table、column、where
password：密码
username：用户名
where(可选)：筛选条件
splitPK（可选）：数据分片字段，一般为主键
writer配置如下
column：写入的字段，其中name指定字段名，type指定类型
compress（可选）：压缩文件类型，不写即没有压缩
defaultFS：hdfs文件系统namenode节点地址，格式：hdfs://ip:端口
fieldDelimiter：字段分隔符
fileName：写入文件名
fileType：文件的类型，只支持text或orc
path：存储到Hadoop hdfs文件系统的路径信息
writeMode： 写入数据处理模式
包括：append（追加）、nonConflict（新增）
编辑配置文件
[[email protected] datax]# vi job/mysql2hdfs.json
添加如下配置
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"column": [
"sensor",
"temp"
],
"connection": [
{
"jdbcUrl": [
"jdbc:mysql://192.168.56.20:3306/mytest"
],
"table": [
"temperatures"
]
}
],
"password": "1234",
"username": "root",
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"column": [
{
"name":"sensor",
"type":"string"
},
{
"name":"temp",
"type":"double"
}
],
"defaultFS": "hdfs://192.168.56.20:9000",
"fieldDelimiter": "\t",
"fileName": "temperatures",
"fileType": "text",
"path": "/mydataxInfo/",
"writeMode": "append"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
运行
[[email protected] datax]# bin/datax.py job/mysql2hdfs.json
使用的fileName可能会自动字hdfs生成各种后缀，我们直接以通配符‘*’来查看
如下所示，同步成功！
3. 读取HDFS数据存放到MySQL
使用如下命令可以查看官方配置模板
[[email protected] datax]# bin/datax.py -r hdfsreader -w mysql mysqlwriter
修改文件名，方便配置
[[email protected] datax]# hdfs dfs -mv /mydataxInfo/temperatures* /mydataxInfo/mytemp.txt
复制导入的新表
mysql> create table mytemp like temperatures;
mysqlreader部分参数说明
writeMode:insert/replace/update(新增、替代、更新)
preSql：在执行到每个表写入数据前，会先执行对应的sql语句
编辑配置文件
[[email protected] datax]# vi job/hdfs2mysql.json
{
"job": {
"content": [
{
"reader": {
"name": "hdfsreader",
"parameter": {
"column": ["*"],
"defaultFS": "hdfs://192.168.56.20:9000",
"encoding": "UTF-8",
"fieldDelimiter": "\t",
"fileType": "text",
"path": "/mydataxInfo/mytemp.txt"
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"column": [
"sensor",
"temp"
],
"connection": [
{
"jdbcUrl": "jdbc:mysql://192.168.56.20:3306/mytest",
"table": ["mytemp"]
}
],
"password": "1234",
"preSql": [],
"session": [],
"username": "root",
"writeMode": "insert"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
执行
[[email protected] datax]# bin/datax.py job/hdfs2mysql.json
如下所示同步成功。
查询执行结果
4. 读取Oralce数据存放到Mysql
将scott练习库中的DEPT表存放到mysql
使用如下命令可以查看官方配置模板
[[email protected] datax]# bin/datax.py -r oraclereader -w mysqlwriter
mysql创建同oracle的表
mysql> create table dept (DEPTNO int,DNAME varchar(15),LOC varchar(15));
编辑配置文件
[[email protected] datax]# vi job/oracle2mysql.json
{
"job": {
"content": [
{
"reader": {
"name": "oraclereader",
"parameter": {
"column": ["*"],
"connection": [
{
"jdbcUrl": ["jdbc:oracle:thin:@//192.168.5.136:1521/ORCL"],
"table": ["dept"]
}
],
"password": "tiger",
"username": "SCOTT"
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"column": ["*"],
"connection": [
{
"jdbcUrl": "jdbc:mysql://192.168.56.20:3306/mytest",
"table": ["dept"]
}
],
"password": "1234",
"preSql": [],
"session": [],
"username": "root",
"writeMode": "insert"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
执行
[[email protected] datax]# bin/datax.py job/oracle2mysql.json
同步成功！
查询同步结果
对于参数以及各种reader和writer仍可以参考官网示例：https://github.com/alibaba/DataX