原来还可以使用 DataX 进行数据同步

刺骨的言语ヽ痛彻心扉 2022-09-12 14:59 227阅读 0赞

### DataX 进行数据同步 ###

*   *   *  DataX
         *   *  一、环境要求
             *  二、DataX部署
             *  三、新建 Job

### DataX ###

DataX 是阿里云 DataWorks 的开源版本，在阿里巴巴集团内被广泛使用的离线数据同步工具/平台。DataX 实现了包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、Hologres、DRDS 等各种异构数据源之间高效的数据同步功能。地址：https://github.com/alibaba/DataX

#### 一、环境要求 ####

*  Python 2.6.X

#### 二、DataX部署 ####

下载解压即可使用。[点击下载][Link 1]

#### 三、新建 Job ####

DataX 的 Job 是以 `Json` 文件方式进行定义，各数据源读写配置详细见地址： https://github.com/alibaba/DataX/wiki/DataX-all-data-channels。

本文以 MaxCompute 同步至 Clickhouse 为例：

1.  安装 Clickhouse 驱动。将 Clickhouse jar 放置目录 `plugin\writer\rdbmswriter\lib`下，同时注册Clickhouse 驱动，修改`datax\plugin\writer\rdbmswriter` 下的 plugin.json 文件，如下所示：
    
        { 
            "name": "rdbmswriter",
            "class": "com.alibaba.datax.plugin.reader.rdbmswriter.RdbmsWriter",
            "description": "useScene: prod. mechanism: Jdbc connection using the database, execute select sql, retrieve data from the ResultSet. warn: The more you know about the database, the less problems you encounter.",
            "developer": "alibaba",
            "drivers":[
        		"ru.yandex.clickhouse.ClickHouseDriver"
        	]
        }
2.  定义 Job
    
    Job 文件可以定义任何路径下，我这是在 DataX 的 Job 目录下创建了 job.json 文件，内容如下：
    
        { 
            "job": { 
                "setting": { 
                    "speed": { 
                        "channel":5
                    },
                    "errorLimit": { 
                        "record": 0,
                        "percentage": 0.02
                    }
                },
                "content": [
                    { 
                        "reader": { 
                            "name": "odpsreader",
                            "parameter": { 
                                "accessId": "accessId",
                                "accessKey": "accessKey",
                                "project": "projectName",
                                "table": "tableName",
                                "column": [
                                    "*"
                                ],
                                "packageAuthorizedProject": "projectName",
                                "splitMode": "record",
                                "odpsServer": "http://xxx/api",
                                "tunnelServer": ""
                            }
                        },
                        "writer": { 
                            "name": "rdbmswriter",
                            "parameter": { 
                                "connection": [
                                    { 
                                        "jdbcUrl": "jdbc:clickhouse://xx.xx.xxx.xxx:8123/databaseName",
                                        "table": [
                                            "tableName"
                                        ]
                                    }
                                ],
                                "username": "username",
                                "password": "password",
                                "table": "tableName",
                                "column": [
                                    "*"
                                ],
        						"preSql": [
        							"TRUNCATE table databaseName.tableName;"
        						]
                            }
                        }
                    }
                ]
            }
        }

上述 Json 大概意思是将 `MaxCompute` 中表 `tableName` 同步到 `Clickhouse` 并在同步前清空表，最大并发数为 5。如果同步过程中出现异常终止同步。

1.  启动 DataX
    
    使用 Python 命令执行同步
    
        cd { YOUR_DATAX_DIR}
        python ../bin/datax.py ../job/job.json

[Link 1]: http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz