DataX的使用

DataX使用指南

1、 DataX3.0概览

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

图1-1 DataX的应用数据库范围

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

2、 DataX3.0的使用

2.1 准备工作

Java版本要求：jdk1.8及以上

参考链接：Java 开发环境配置

Python版本要求：2.7.X，DataX未更新至Python3

参考链接：Python 环境搭建

DataX下载：DataX下载地址

当然你还可以去Github去下载源码进行编译，通过此种方式你可以自行选择你需要的配置进行打包，如只需要Mysql的，那么其他的相关支持你都不需要用到，也就不需要都打包，更轻量使用。与此同时，你还需要maven工具进行打包，在pom.Xml中删除你不需要的模块，再执行maven命令：mvn -U clean package assembly:assembly -Dmaven.test.skip=true，生成的文件在/target/datax/datax/下

确保上述步骤都通过后则继续往下看。

2.2 快速上手

2.2.1 构建一个测试数据表

表结构如表2-1所示。

表2-1 测试表结构

字段名	类型	备注
name	varchar
age	int
age_true	int

并向其中插入40条数据，如表2-2所示。

 表2-2 测试表数据

name	age	age_true
tom	23
tom	23
tom	23
tom	23
tom	23
tom	23
tom	23
tom	23
...	...

2.2.2 构建任务Json

DataX工具是用json文件作为配置文件的，根据官方提供文档我们构建Json文件如下所示。

{ 

    "job": { 

        "content": [ 

            { 

                "reader": { 

                    "name": "mysqlreader",

                    "parameter": { 

                        "username": "root", 

                        "password": "123456",  

                        "column": ["name","age"], 

                        "where": "age<100", 

                        "connection": [ 

                            { 

                                "table": [ 

                                    "person" 

                                ], 

                                "jdbcUrl": [ 

                                    "jdbc:mysql://127.0.0.1:3306/test?characterEncoding=utf8"

                                ] 

                            } 

                        ] 

                    } 

                }, 

                "writer": { 

                    "name": "mysqlwriter",

                    "parameter": { 

                        "username": "root", 

                        "password": "123456", 

                        "column": ["name","age_true"],

                        "connection": [ 

                            { 

                                "table": [ 

                                    "person" 

                                ], 

                                "jdbcUrl":"jdbc:mysql://127.0.0.1:3306/test?characterEncoding=utf8"                              

                            } 

                        ]   

                    } 

                } 

            } 

        ], 

        "setting": {

                 "speed": {

                         "channel": 1,

                         "byte": 104857600

                 },

                 "errorLimit": {

                         "record": 10,

                         "percentage": 0.05

      }

    } 

}

它由三部分组成，分别是读，写和通用配置。

Reader部分，也就是读，常用以下几种参数，如表2-3所示。

表2-3 读参数表

参数名	解释	备注
name	与要读取的数据库一致	字符串
jdbcUrl	数据库链接	数组 会自动选择一个合法的链接 可以填写连接附件控制信息
username	用户名	字符串，数据库的用户名
password	密码	字符串，数据库的密码
table	要同步的表名	数组，需保证表结构一致
column	要同步的列名	数组
where	选取的条件	字符串
querySql	自定义查询语句	会自动忽略上述的同步条件

Writer部分，也就是写，常用以下几种参数，如表2-4所示。

表2-4 写参数表

参数名	解释	备注
name	与要读取的数据库一致	字符串
jdbcUrl	数据库链接	字符串 不和writer一样 可以填写连接附件控制信息
username	用户名	字符串，数据库的用户名
password	密码	字符串，数据库的密码
table	要同步的表名	数组，需保证表结构一致
column	要同步的列名	列名可以不对应，但是类型和总的个数要一致
preSql	写入前执行的语句	数组，比如清空表等
postSql	写入后执行的语句	数组
writeMode	写入方式，默认为insert	insert/replace/update

job.setting.speed(流量控制)

Job支持用户对速度的自定义控制，channel的值可以控制同步时的并发数，byte的值可以控制同步时的速度

job.setting.errorLimit(脏数据控制)

Job支持用户对于脏数据的自定义监控和告警，包括对脏数据最大记录数阈值（record值）或者脏数据占比阈值（percentage值），当Job传输过程出现的脏数据大于用户指定的数量/百分比，DataX Job报错退出。

2.2.3 执行

Win+R+cmd进入命令行控制台，首先通过java –version和python查看是否满足要求，然后执行：python 空格{datax文件夹路径}\bin\datax.py 空格{json配置文件的路径}，如，python D:\download\datax\bin\datax.py D:\download\datax\job\job2.json

接着控制台会打印出相应的信息，控制台乱码输入 chcp空格 65001，如图2-1所示。

图2-1 命令行打印输出信息

我们可以看到写入数据表中的数据已经发生了变化，此处是将age位置的信息写入到age_true位置上，结果如表2-5所示。

表2-5 DataX写入结果

name	age	age_true
tom	23
tom	23
tom	23
tom	23
tom		23
tom		23
tom		23
tom		23
...	...

2.3 java项目中使用

数据库迁移通常是定期的，所以一般情况下我们是将他用作定时任务的，所以，我们需要在java环境下用定时任务去执行。

大致思路如下：

首先获得指定文件夹下的任务配置文件，然后调用cmd执行cmd程序，具体实现如下：

声明参数，参数请根据实际去修改：

//datax.py文件的路径

    @Value(value = "D:\\download\\datax\\bin\\datax.py")

    private String dataxPath;

    //任务文件夹的路径

    @Value(value = "D:\\download\\datax\\job\\")

    private String jsonPath;

    //python路径

    @Value(value = "C:\\Program Files\\Python27\\python.exe")

    private String pythonPath;

//获得任务文件夹下的所有json文件

    public File[] getFileList(){

        File file=new File(jsonPath);

        File[] files=file.listFiles((File f)->f.getName().endsWith(".json"));

        return files;

    }

//主程序

public void doTask(){

        File[] files=getFileList();

        for(File f:files){

            String cmd = pythonPath+" "+dataxPath+" "+f.getAbsolutePath();

            try {

                Process process = Runtime.getRuntime().exec(cmd);

                //返回信息写入流用控制台打出来

//此处转码，不然控制台中文乱码

                BufferedReader in = new BufferedReader(new InputStreamReader(process.getInputStream(),"utf-8"));

                String line = null;

                while ((line = in.readLine()) != null) {

                    System.out.println(line);

                }

                in.close();

                process.waitFor();

            }

            catch (Exception e){

                logger.error(e.getMessage());

            }

        }

    }

至此，DataX3.0的使用已经介绍完成，其余内容只需作修改即可。

3、 相关链接

下面列出常用数据库的读和写说明文档，更多请访问DataX的GitHub地址

MysqlReader 插件文档

MysqlWriter 插件文档

OracleReader 插件文档

OracleWriter插件文档