DataX入门教程

概述

什么是DataX

​ DataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。

DataX的设计

为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。

支持的数据源

DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入。

框架设计

运行原理

举例来说，用户提交了一个DataX作业，并且配置了20个并发，目的是将一个100张分表的mysql数据同步到odps里面。 DataX的调度决策思路是：

1）DataXJob根据分库分表切分成了100个Task。

2）根据20个并发，DataX计算共需要分配4个TaskGroup。

3）4个TaskGroup平分切分好的100个Task，每一个TaskGroup负责以5个并发共计运行25个Task。

与Sqoop的对比

快速入门

官方地址

下载地址：http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz

源码地址：https://github.com/alibaba/DataX

前置要求

- Linux

- JDK(1.8以上，推荐1.8)

- Python(推荐Python2.6.X)

安装

1）将下载好的datax.tar.gz上传到hadoop102的/opt/software

2）解压datax.tar.gz到/opt/module

[daydayup@hadoop102 software]$ tar -zxvf datax.tar.gz -C /opt/module/

3）运行自检脚本

[daydayup@hadoop102 bin]$ cd /opt/module/datax/bin/

[daydayup@hadoop102 bin]$ python datax.py /opt/module/datax/job/job.json

使用案例

从stream流读取数据并打印到控制台

1）查看配置模板

[daydayup@hadoop102 bin]$ python datax.py -r streamreader -w streamwriter

DataX (DATAX-OPENSOURCE-3.0), From Alibaba !

Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.

Please refer to the streamreader document:

https://github.com/alibaba/DataX/blob/master/streamreader/doc/streamreader.md

Please refer to the streamwriter document:

https://github.com/alibaba/DataX/blob/master/streamwriter/doc/streamwriter.md

Please save the following configuration as a json file and use

python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json

to run the job.

{

"job": {

"content": [

{

"reader": {

"name": "streamreader",

"parameter": {

"column": [],

"sliceRecordCount": ""

}

},

"writer": {

"name": "streamwriter",

"parameter": {

"encoding": "",

"print": true

}

}

}

],

"setting": {

"speed": {

"channel": ""

}

}

}

}

2）根据模板编写配置文件

[daydayup@hadoop102 job]$ vim stream2stream.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": 1

}

}

}

}

3）运行

[daydayup@hadoop102 job]$

/opt/module/datax/bin/datax.py /opt/module/datax/job/stream2stream.json

读取MySQL中的数据存放到HDFS

查看官方模板

[daydayup@hadoop102 ~]$ python /opt/module/datax/bin/datax.py -r mysqlreader -w hdfswriter

{

"job": {

"content": [

{

"reader": {

"name": "mysqlreader",

"parameter": {

"column": [],

"connection": [

{

"jdbcUrl": [],

"table": []

}

],

"password": "",

"username": "",

"where": ""

}

},

"writer": {

"name": "hdfswriter",

"parameter": {

"column": [],

"compress": "",

"defaultFS": "",

"fieldDelimiter": "",

"fileName": "",

"fileType": "",

"path": "",

"writeMode": ""

}

}

}

],

"setting": {

"speed": {

"channel": ""

}

}

}

}

mysqlreader参数解析：

hdfswriter参数解析：

准备数据

1）创建student表

mysql> create database datax;

mysql> use datax;

mysql> create table student(id int,name varchar(20));

2）插入数据

mysql> insert into student values(1001,'zhangsan'),(1002,'lisi'),(1003,'wangwu');

编写配置文件

[daydayup@hadoop102 datax]$ vim /opt/module/datax/job/mysql2hdfs.json

{

"job": {

"content": [

{

"reader": {

"name": "mysqlreader",

"parameter": {

"column": [

"id",

"name"

],

"connection": [

{

"jdbcUrl": [

"jdbc:mysql://hadoop102:3306/datax"

],

"table": [

"student"

]

}

],

"username": "root",

"password": "000000"

}

},

"writer": {

"name": "hdfswriter",

"parameter": {

"column": [

{

"name": "id",

"type": "int"

},

{

"name": "name",

"type": "string"

}

],

"defaultFS": "hdfs://hadoop102:9000",

"fieldDelimiter": "\t",

"fileName": "student.txt",

"fileType": "text",

"path": "/",

"writeMode": "append"

}

}

}

],

"setting": {

"speed": {

"channel": "1"

}

}

}

}

执行任务

[daydayup@hadoop102 datax]$ bin/datax.py job/mysql2hdfs.json

2019-05-17 16:02:16.581 [job-0] INFO JobContainer -

任务启动时刻 : 2019-05-17 16:02:04

任务结束时刻 : 2019-05-17 16:02:16

任务总计耗时 : 12s

任务平均流量 : 3B/s

记录写入速度 : 0rec/s

读出记录总数 : 3

读写失败总数 : 0

查看hdfs

注意：HdfsWriter实际执行时会在该文件名后添加随机的后缀作为每个线程写入实际文件名。

关于HA的支持

"hadoopConfig":{
"dfs.nameservices": "ns",
"dfs.ha.namenodes.ns": "nn1,nn2",
"dfs.namenode.rpc-address.ns.nn1": "主机名:端口",
"dfs.namenode.rpc-address.ns.nn2": "主机名:端口",
"dfs.client.failover.proxy.provider.ns": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
}

读取HDFS数据写入MySQL

1）将上个案例上传的文件改名

[daydayup@hadoop102 datax]$ hadoop fs -mv /student.txt* /student.txt

2）查看官方模板

[daydayup@hadoop102 datax]$ python bin/datax.py -r hdfsreader -w mysqlwriter

{

"job": {

"content": [

{

"reader": {

"name": "hdfsreader",

"parameter": {

"column": [],

"defaultFS": "",

"encoding": "UTF-8",

"fieldDelimiter": ",",

"fileType": "orc",

"path": ""

}

},

"writer": {

"name": "mysqlwriter",

"parameter": {

"column": [],

"connection": [

{

"jdbcUrl": "",

"table": []

}

],

"password": "",

"preSql": [],

"session": [],

"username": "",

"writeMode": ""

}

}

}

],

"setting": {

"speed": {

"channel": ""

}

}

}

}

3）创建配置文件

[daydayup@hadoop102 datax]$ vim job/hdfs2mysql.json

{

"job": {

"content": [

{

"reader": {

"name": "hdfsreader",

"parameter": {

"column": ["*"],

"defaultFS": "hdfs://hadoop102:9000",

"encoding": "UTF-8",

"fieldDelimiter": "\t",

"fileType": "text",

"path": "/student.txt"

}

},

"writer": {

"name": "mysqlwriter",

"parameter": {

"column": [

"id",

"name"

],

"connection": [

{

"jdbcUrl": "jdbc:mysql://hadoop102:3306/datax",

"table": ["student2"]

}

],

"password": "000000",

"username": "root",

"writeMode": "insert"

}

}

}

],

"setting": {

"speed": {

"channel": "1"

}

}

}

}

4）在MySQL的datax数据库中创建student2

mysql> use datax;

mysql> create table student2(id int,name varchar(20));

5）执行任务

[daydayup@hadoop102 datax]$ bin/datax.py job/hdfs2mysql.json

2019-05-17 16:21:53.616 [job-0] INFO JobContainer -

任务启动时刻 : 2019-05-17 16:21:41

任务结束时刻 : 2019-05-17 16:21:53

任务总计耗时 : 11s

任务平均流量 : 3B/s

记录写入速度 : 0rec/s

读出记录总数 : 3

读写失败总数 : 0

6）查看student2表

mysql> select * from student2;

+------+----------+

| id | name |

+------+----------+

| 1001 | zhangsan |

| 1002 | lisi |

| 1003 | wangwu |

+------+----------+

3 rows in set (0.00 sec)

Oracle数据库

以下操作使用root账号。

oracle数据库简介

Oracle Database，又名Oracle RDBMS，或简称Oracle。是甲骨文公司的一款关系数据库管理系统。它是在数据库领域一直处于领先地位的产品。可以说Oracle数据库系统是目前世界上流行的关系数据库管理系统，系统可移植性好、使用方便、功能强，适用于各类大、中、小、微机环境。它是一种高效率、可靠性好的、适应高吞吐量的数据库解决方案。

安装前的准备

安装依赖

yum install -y bc binutils compat-libcap1 compat-libstdc++33 elfutils-libelf elfutils-libelf-devel fontconfig-devel glibc glibc-devel ksh libaio libaio-devel libX11 libXau libXi libXtst libXrender libXrender-devel libgcc libstdc++ libstdc++-devel libxcb make smartmontools sysstat kmod* gcc-c++ compat-libstdc++-33

配置用户组

Oracle安装文件不允许通过root用户启动，需要为oracle配置一个专门的用户。

1）创建sql 用户组

[root@hadoop102 software]#groupadd sql

2）创建oracle 用户并放入sql组中

[root@hadoop102 software]#useradd oracle -g sql

3）修改oracle用户登录密码，输入密码后即可使用oracle用户登录系统

[root@hadoop102 software]#passwd oracle

上传安装包并解压

注意:19c需要把软件包直接解压到ORACLE_HOME的目录下

[root@hadoop102 software]# mkdir -p /home/oracle/app/oracle/product/19.3.0/dbhome_1

[root@hadoop102 software]# unzip LINUX.X64_193000_db_home.zip -d /home/oracle/app/oracle/product/19.3.0/dbhome_1

修改所属用户和组

[root@hadoop102 dbhome_1]# chown -R oracle:sql /home/oracle/app/

修改配置文件sysctl.conf

[root@hadoop102 module]# vim /etc/sysctl.conf

删除里面的内容，添加如下内容：

net.ipv4.ip_local_port_range = 9000 65500

fs.file-max = 6815744

kernel.shmall = 10523004

kernel.shmmax = 6465333657

kernel.shmmni = 4096

kernel.sem = 250 32000 100 128

net.core.rmem_default=262144

net.core.wmem_default=262144

net.core.rmem_max=4194304

net.core.wmem_max=1048576

fs.aio-max-nr = 1048576

参数解析：

net.ipv4.ip_local_port_range ：可使用的IPv4端口范围

fs.file-max ：该参数表示文件句柄的最大数量。文件句柄设置表示在linux系统中可以打开的文件数量。

kernel.shmall ：该参数表示系统一次可以使用的共享内存总量（以页为单位）

kernel.shmmax ：该参数定义了共享内存段的最大尺寸（以字节为单位）

kernel.shmmni ：这个内核参数用于设置系统范围内共享内存段的最大数量

kernel.sem ： 该参数表示设置的信号量。

net.core.rmem_default：默认的TCP数据接收窗口大小（字节）。

net.core.wmem_default：默认的TCP数据发送窗口大小（字节）。

net.core.rmem_max：最大的TCP数据接收窗口（字节）。

net.core.wmem_max：最大的TCP数据发送窗口（字节）。

fs.aio-max-nr ：同时可以拥有的的异步IO请求数目。

修改配置文件limits.conf

[root@hadoop102 module]# vim /etc/security/limits.conf

在文件末尾添加：

oracle soft nproc 2047

oracle hard nproc 16384

oracle soft nofile 1024

oracle hard nofile 65536

重启机器生效。

安装Oracle数据库

设置环境变量

[oracle@hadoop102 dbhome_1]# vim /home/oracle/.bash_profile

添加：

#ORACLE_HOME

export ORACLE_BASE=/home/oracle/app/oracle

export ORACLE_HOME=/home/oracle/app/oracle/product/19.3.0/dbhome_1

export PATH=$PATH:$ORACLE_HOME/bin

export ORACLE_SID=orcl

export NLS_LANG=AMERICAN_AMERICA.ZHS16GBK

[oracle@hadoop102 ~]$ source /home/oracle/.bash_profile

进入虚拟机图像化页面操作

[oracle@hadoop102 ~]# cd /opt/module/oracle

[oracle@hadoop102 database]# ./runInstaller

安装数据库

1）选择仅安装数据库软件

2）选择单实例数据库安装

3）选择企业版，默认

4）设置安装位置

5）操作系统组设置

6）配置root脚本自动执行

7）条件检查通过后，选择开始安装

8）运行root脚本

9）安装完成

设置Oracle监听

命令行输入以下命令

[oracle@hadoop2 ~]$ netca

选择添加

设置监听名，默认即可

选择协议，默认即可

设置端口号，默认即可

配置更多监听，默认

完成

创建数据库

进入创建页面

[oracle@hadoop2 ~]$ dbca

选择创建数据库

选择高级配置

选择数据仓库

将图中所示对勾去掉

存储选项

快速恢复选项

选择监听程序

如图设置

使用自动内存管理

管理选项，默认

设置统一密码

创建选项，选择创建数据库

概要，点击完成

等待安装

简单使用

开启，关闭监听服务

开启服务：

[oracle@hadoop102 ~]$ lsnrctl start

关闭服务：

[oracle@hadoop102 ~]$ lsnrctl stop

进入命令行

[oracle@hadoop102 ~]$ sqlplus

SQL*Plus: Release 19.0.0.0.0 - Production on Fri Sep 3 01:44:30 2021

Version 19.3.0.0.0

Copyright (c) 1982, 2019, Oracle. All rights reserved.

Enter user-name: system

Enter password: （这里输入之前配置的统一密码）

Connected to:

Oracle Database 19c Enterprise Edition Release 19.0.0.0.0 - Production

Version 19.3.0.0.0

SQL>

创建用户并授权

SQL> create user daydayup identified by 000000;

User created.

SQL> grant create session,create table,create view,create sequence,unlimited tablespace to daydayup;

Grant succeeded.

进入daydayup账号，创建表

SQL>create TABLE student(id INTEGER,name VARCHAR2(20));

SQL>insert into student values (1,'zhangsan');

SQL> select * from student;

ID NAME

---------- ----------------------------------------

1 zhangsan

注意：安装完成后重启机器可能出现ORACLE not available错误，解决方法如下：

[oracle@hadoop102 ~]$ sqlplus / as sysdba

SQL>startup

SQL>conn daydayup

Enter password:

Oracle与MySQL的SQL区别

DataX案例

从Oracle中读取数据存到MySQL

1）MySQL中创建表

[oracle@hadoop102 ~]$ mysql -uroot -p000000

mysql> create database oracle;

mysql> use oracle;

mysql> create table student(id int,name varchar(20));

2）编写datax配置文件

[oracle@hadoop102 ~]$ vim /opt/module/datax/job/oracle2mysql.json

{

"job": {

"content": [

{

"reader": {

"name": "oraclereader",

"parameter": {

"column": ["*"],

"connection": [

{

"jdbcUrl": ["jdbc:oracle:thin:@hadoop102:1521:orcl"],

"table": ["student"]

}

],

"password": "000000",

"username": "daydayup"

}

},

"writer": {

"name": "mysqlwriter",

"parameter": {

"column": ["*"],

"connection": [

{

"jdbcUrl": "jdbc:mysql://hadoop102:3306/oracle",

"table": ["student"]

}

],

"password": "000000",

"username": "root",

"writeMode": "insert"

}

}

}

],

"setting": {

"speed": {

"channel": "1"

}

}

}

}

3）执行命令

[oracle@hadoop102 ~]$

/opt/module/datax/bin/datax.py /opt/module/datax/job/oracle2mysql.json

查看结果:

mysql> select * from student;

+------+----------+

| id | name |

+------+----------+

| 1 | zhangsan |

+------+----------+

读取Oracle的数据存入HDFS中

1）编写配置文件

[oracle@hadoop102 datax]$ vim job/oracle2hdfs.json

{

"job": {

"content": [

{

"reader": {

"name": "oraclereader",

"parameter": {

"column": ["*"],

"connection": [

{

"jdbcUrl": ["jdbc:oracle:thin:@hadoop102:1521:orcl"],

"table": ["student"]

}

],

"password": "000000",

"username": "daydayup"

}

},

"writer": {

"name": "hdfswriter",

"parameter": {

"column": [

{

"name": "id",

"type": "int"

},

{

"name": "name",

"type": "string"

}

],

"defaultFS": "hdfs://hadoop102:9000",

"fieldDelimiter": "\t",

"fileName": "oracle.txt",

"fileType": "text",

"path": "/",

"writeMode": "append"

}

}

}

],

"setting": {

"speed": {

"channel": "1"

}

}

}

}

2）执行

[oracle@hadoop102 datax]$ bin/datax.py job/oracle2hdfs.json

3）查看HDFS结果