DataX使用

预备

DataX介绍

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

设计理念

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

DataX3.0框架设计

DataX本身作为离线数据同步框架，采用Framework + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。
Reader：Reader为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
Writer： Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
Framework：Framework用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。

Github下载地址：https://github.com/alibaba/DataX#quick-start

环境准备

JDK1.8

Python2.x (3.x会导致语法错误，可以修改后使用)

2.1 cmd窗口中输入CHCP 65001 防止中文乱码
2.2 python3出现语法错误是因为 /bin 目录下 py文件的print没有(),可以自行修改

测试

解压下载的DataX后测试

cd %DATAX_HOME%\bin
# %DATAX_HOME% 为你自己的解压路径

# 查看mysql json文件模板
python datax.py -r mysqlreader -w mysqlmwriter

# 可以替换参数查看对应数据库的json模板
# 或者直接在 %DATAX_HOME%\plugin 中查看

编辑自己的mysql的json模板
vim %DATAX_HOME%\job\mysql2mysql.json

{
    "job": {
        "content": [
            {
                  "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": ["id","name"],  // "*"默认所有列,支持列裁剪，列换序
                        "connection": [
                            {
                                // 也可以使用自定义sql语句获取需要读取的值
                                // "querySql": ["SELECT id,name FROM student where id>'5000' ;"],
                                "jdbcUrl": ["jdbc:mysql://localhost:3306/test1?useUnicode=true&characterEncoding=utf8&yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true"],
                                "table": ["student"]  // 可以支持多张表同时抽取
                            }
                        ],
                        "password": "*****",
                        "username": "*****",
                        
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                           "column": ["id","name"],
                        "connection": [
                            {
                                
                                "jdbcUrl": "jdbc:mysql://localhost/test",
                                "table": ["student"]
                            }
                        ],
                        "password": "*****",
                        "username": "*****",
                        "preSql": [], // 写入数据到目的表前，会先执行这里的标准语句。
                        "postSql":[], // 写入数据到目的表后，会执行这里的标准语句。
                        "session": [], // DataX在获取Mysql连接时，执行session指定的SQL语句，修改当前connection session属性
                        "writeMode": ""  // 默认insert ，可选insert/replace/update 
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }
}

执行 mysql2mysql.json

python %DATAX_HOME%/bin/datax.py %DATAX_HOME%\job\mysql2mysql.json

欢迎大家留言，一起讨论。

最后编辑于：2021.05.06 15:21:16