前言

我們公司有個項目的數(shù)據(jù)量高達五千萬，但是因為報表那塊數(shù)據(jù)不太準確，業(yè)務(wù)庫和報表庫又是跨庫操作，所以并不能使用 SQL 來進行同步。當時的打算是通過 mysqldump 或者存儲的方式來進行同步，但是嘗試后發(fā)現(xiàn)這些方案都不切實際：

mysqldump：不僅備份需要時間，同步也需要時間，而且在備份的過程，可能還會有數(shù)據(jù)產(chǎn)出（也就是說同步等于沒同步）

存儲方式：這個效率太慢了，要是數(shù)據(jù)量少還好，我們使用這個方式的時候，三個小時才同步兩千條數(shù)據(jù)…

后面在網(wǎng)上查看后，發(fā)現(xiàn) DataX 這個工具用來同步不僅速度快，而且同步的數(shù)據(jù)量基本上也相差無幾。

一、DataX 簡介

DataX 是阿里云 DataWorks 數(shù)據(jù)集成 的開源版本，主要就是用于實現(xiàn)數(shù)據(jù)間的離線同步。 DataX 致力于實現(xiàn)包括關(guān)系型數(shù)據(jù)庫（MySQL、Oracle 等）、HDFS、Hive、ODPS、HBase、FTP 等 各種異構(gòu)數(shù)據(jù)源（即不同的數(shù)據(jù)庫） 間穩(wěn)定高效的數(shù)據(jù)同步功能。

為了 解決異構(gòu)數(shù)據(jù)源同步問題，DataX 將復雜的網(wǎng)狀同步鏈路變成了星型數(shù)據(jù)鏈路 ，DataX 作為中間傳輸載體負責連接各種數(shù)據(jù)源；

當需要接入一個新的數(shù)據(jù)源時，只需要將此數(shù)據(jù)源對接到 DataX，便能跟已有的數(shù)據(jù)源作為無縫數(shù)據(jù)同步。

1.DataX3.0 框架設(shè)計

DataX 采用 Framework + Plugin 架構(gòu)，將數(shù)據(jù)源讀取和寫入抽象稱為 Reader/Writer 插件，納入到整個同步框架中。

2.DataX3.0 核心架構(gòu)

DataX 完成單個數(shù)據(jù)同步的作業(yè)，我們稱為 Job，DataX 接收到一個 Job 后，將啟動一個進程來完成整個作業(yè)同步過程。DataX Job 模塊是單個作業(yè)的中樞管理節(jié)點，承擔了數(shù)據(jù)清理、子任務(wù)切分、TaskGroup 管理等功能。

DataX Job 啟動后，會根據(jù)不同源端的切分策略，將 Job 切分成多個小的 Task (子任務(wù))，以便于并發(fā)執(zhí)行。

接著 DataX Job 會調(diào)用 Scheduler 模塊，根據(jù)配置的并發(fā)數(shù)量，將拆分成的 Task 重新組合，組裝成 TaskGroup（任務(wù)組）

每一個 Task 都由 TaskGroup 負責啟動，Task 啟動后，會固定啟動 Reader --> Channel --> Writer 線程來完成任務(wù)同步工作。

DataX 作業(yè)運行啟動后，Job 會對 TaskGroup 進行監(jiān)控操作，等待所有 TaskGroup 完成后，Job 便會成功退出（異常退出時 值非 0 ）

DataX 調(diào)度過程：

首先 DataX Job 模塊會根據(jù)分庫分表切分成若干個 Task，然后根據(jù)用戶配置并發(fā)數(shù)，來計算需要分配多少個 TaskGroup；

計算過程：Task / Channel = TaskGroup，最后由 TaskGroup 根據(jù)分配好的并發(fā)數(shù)來運行 Task（任務(wù)）

二、使用 DataX 實現(xiàn)數(shù)據(jù)同步

準備工作：

JDK（1.8 以上，推薦 1.8）

Python（2，3 版本都可以）

Apache Maven 3.x（Compile DataX）（手動打包使用，使用 tar 包方式不需要安裝）

安裝 JDK：

需要創(chuàng)建 Oracle 賬號

[root@MySQL-1 ~]# ls
anaconda-ks.cfg  jdk-8u181-linux-x64.tar.gz
[root@MySQL-1 ~]# tar zxf jdk-8u181-linux-x64.tar.gz 
[root@DataX ~]# ls
anaconda-ks.cfg  jdk1.8.0_181  jdk-8u181-linux-x64.tar.gz
[root@MySQL-1 ~]# mv jdk1.8.0_181 /usr/local/java
[root@MySQL-1 ~]# cat <> /etc/profile
export JAVA_HOME=/usr/local/java
export PATH=$PATH:"$JAVA_HOME/bin"
END
[root@MySQL-1 ~]# source /etc/profile
[root@MySQL-1 ~]# java -version

因為 CentOS 7 上自帶 Python 2.7 的軟件包，所以不需要進行安裝。

1.Linux 上安裝 DataX 軟件

[root@MySQL-1 ~]# wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
[root@MySQL-1 ~]# tar zxf datax.tar.gz -C /usr/local/
[root@MySQL-1 ~]# rm -rf /usr/local/datax/plugin/*/._*# 需要刪除隱藏文件 (重要)

當未刪除時，可能會輸出：[/usr/local/datax/plugin/reader/._drdsreader/plugin.json] 不存在. 請檢查您的配置文件.

驗證：

[root@MySQL-1 ~]# cd /usr/local/datax/bin
[root@MySQL-1 ~]# python datax.py ../job/job.json# 用來驗證是否安裝成功

輸出：

2021-12-13 1928.828 [job-0] INFO  JobContainer - PerfTrace not enable!
2021-12-13 1928.829 [job-0] INFO  StandAloneJobContainerCommunicator - Total 100000 records, 2600000 bytes | Speed 253.91KB/s, 10000 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.060s |  All Task WaitReaderTime 0.068s | Percentage 100.00%
2021-12-13 1928.829 [job-0] INFO  JobContainer - 
任務(wù)啟動時刻                    : 2021-12-13 1918
任務(wù)結(jié)束時刻                    : 2021-12-13 1928
任務(wù)總計耗時                    :                 10s
任務(wù)平均流量                    :          253.91KB/s
記錄寫入速度                    :          10000rec/s
讀出記錄總數(shù)                    :              100000
讀寫失敗總數(shù)                    :                   0

2.DataX 基本使用

查看 streamreader --> streamwriter 的模板：

[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py -r streamreader -w streamwriter

輸出：

DataX(DATAX-OPENSOURCE-3.0),FromAlibaba!
Copyright(C)2010-2017,AlibabaGroup.AllRightsReserved.


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

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

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

{
"job":{
"content":[
{
"reader":{
"name":"streamreader",
"parameter":{
"column":[],
"sliceRecordCount":""
}
},
"writer":{
"name":"streamwriter",
"parameter":{
"encoding":"",
"print":true
}
}
}
],
"setting":{
"speed":{
"channel":""
}
}
}
}

根據(jù)模板編寫 json 文件

[root@MySQL-1~]#cat<test.json
{
"job":{
"content":[
{
"reader":{
"name":"streamreader",
"parameter":{
"column":[#同步的列名(*表示所有)
{
"type":"string",
"value":"Hello."
},
{
"type":"string",
"value":"河北彭于晏"
},
],
"sliceRecordCount":"3"#打印數(shù)量
}
},
"writer":{
"name":"streamwriter",
"parameter":{
"encoding":"utf-8",#編碼
"print":true
}
}
}
],
"setting":{
"speed":{
"channel":"2"#并發(fā)(即sliceRecordCount*channel=結(jié)果)
}
}
}
}

輸出：（要是復制我上面的話，需要把 # 帶的內(nèi)容去掉）

3.安裝 MySQL 數(shù)據(jù)庫

分別在兩臺主機上安裝：

[root@MySQL-1 ~]# yum -y install mariadb mariadb-server mariadb-libs mariadb-devel   
[root@MySQL-1 ~]# systemctl start mariadb# 安裝 MariaDB 數(shù)據(jù)庫
[root@MySQL-1 ~]# mysql_secure_installation# 初始化
NOTE: RUNNING ALL PARTS OF THIS SCRIPT IS RECOMMENDED FOR ALL MariaDB
      SERVERS IN PRODUCTION USE!  PLEASE READ EACH STEP CAREFULLY!

Enter current password for root (enter for none):     # 直接回車
OK, successfully used password, moving on...
Set root password? [Y/n] y                         # 配置 root 密碼
New password: 
Re-enter new password: 
Password updated successfully!
Reloading privilege tables..
 ... Success!
Remove anonymous users? [Y/n] y                 # 移除匿名用戶
 ... skipping.
Disallow root login remotely? [Y/n] n             # 允許 root 遠程登錄
 ... skipping.
Remove test database and access to it? [Y/n] y      # 移除測試數(shù)據(jù)庫
 ... skipping.
Reload privilege tables now? [Y/n] y                  # 重新加載表
 ... Success!

1）準備同步數(shù)據(jù)（要同步的兩臺主機都要有這個表）

MariaDB [(none)]> create database `course-study`;
Query OK, 1 row affected (0.00 sec)

MariaDB [(none)]> create table `course-study`.t_member(ID int,Name varchar(20),Email varchar(30));
Query OK, 0 rows affected (0.00 sec)

因為是使用 DataX 程序進行同步的，所以需要在雙方的數(shù)據(jù)庫上開放權(quán)限：

grant all privileges on *.* to root@'%' identified by '123123';
flush privileges;

2）創(chuàng)建存儲過程：

DELIMITER $$
CREATE PROCEDURE test()
BEGIN
declare A int default 1;
while (A < 3000000)do
insert into `course-study`.t_member values(A,concat("LiSa",A),concat("LiSa",A,"@163.com"));
set A = A + 1;
END while;
END $$
DELIMITER ;

3）調(diào)用存儲過程（在數(shù)據(jù)源配置，驗證同步使用）：

call test();

4.通過 DataX 實 MySQL 數(shù)據(jù)同步

1）生成 MySQL 到 MySQL 同步的模板：

[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py -r mysqlreader -w mysqlwriter
{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",# 讀取端
                    "parameter": {
                        "column": [], # 需要同步的列 (* 表示所有的列)
                        "connection": [
                            {
                                "jdbcUrl": [], # 連接信息
                                "table": []# 連接表
                            }
                        ], 
                        "password": "", # 連接用戶
                        "username": "", # 連接密碼
                        "where": ""# 描述篩選條件
                    }
                }, 
                "writer": {
                    "name": "mysqlwriter",# 寫入端
                    "parameter": {
                        "column": [], # 需要同步的列
                        "connection": [
                            {
                                "jdbcUrl": "", # 連接信息
                                "table": []# 連接表
                            }
                        ], 
                        "password": "", # 連接密碼
                        "preSql": [], # 同步前. 要做的事
                        "session": [], 
                        "username": "",# 連接用戶 
                        "writeMode": ""# 操作類型
                    }
                }
            }
        ], 
        "setting": {
            "speed": {
                "channel": ""# 指定并發(fā)數(shù)
            }
        }
    }
}

2）編寫 json 文件：

[root@MySQL-1 ~]# vim install.json
{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader", 
                    "parameter": {
                        "username": "root",
                        "password": "123123",
                        "column": ["*"],
                        "splitPk": "ID",
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc//192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"
                                ], 
                                "table": ["t_member"]
                            }
                        ]
                    }
                }, 
                "writer": {
                    "name": "mysqlwriter", 
                    "parameter": {
                        "column": ["*"], 
                        "connection": [
                            {
                                "jdbcUrl": "jdbc//192.168.1.2:3306/course-study?useUnicode=true&characterEncoding=utf8",
                                "table": ["t_member"]
                            }
                        ], 
                        "password": "123123",
                        "preSql": [
                            "truncate t_member"
                        ], 
                        "session": [
                            "set session sql_mode='ANSI'"
                        ], 
                        "username": "root", 
                        "writeMode": "insert"
                    }
                }
            }
        ], 
        "setting": {
            "speed": {
                "channel": "5"
            }
        }
    }
}

3）驗證

[root@MySQL-1 ~]# python /usr/local/datax/bin/datax.py install.json

輸出：

2021-12-15 1615.120 [job-0] INFO  JobContainer - PerfTrace not enable!
2021-12-15 1615.120 [job-0] INFO  StandAloneJobContainerCommunicator - Total 2999999 records, 107666651 bytes | Speed 2.57MB/s, 74999 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 82.173s |  All Task WaitReaderTime 75.722s | Percentage 100.00%
2021-12-15 1615.124 [job-0] INFO  JobContainer - 
任務(wù)啟動時刻                    : 2021-12-15 1632
任務(wù)結(jié)束時刻                    : 2021-12-15 1615
任務(wù)總計耗時                    :                 42s
任務(wù)平均流量                    :            2.57MB/s
記錄寫入速度                    :          74999rec/s
讀出記錄總數(shù)                    :             2999999
讀寫失敗總數(shù)                    :                   0

你們可以在目的數(shù)據(jù)庫進行查看，是否同步完成。

上面的方式相當于是完全同步，但是當數(shù)據(jù)量較大時，同步的時候被中斷，是件很痛苦的事情；

所以在有些情況下，增量同步還是蠻重要的。

5.使用 DataX 進行增量同步

使用 DataX 進行全量同步和增量同步的唯一區(qū)別就是：增量同步需要使用 where 進行條件篩選。 （即，同步篩選后的 SQL）

1）編寫 json 文件：

[root@MySQL-1~]#vimwhere.json
{
"job":{
"content":[
{
"reader":{
"name":"mysqlreader",
"parameter":{
"username":"root",
"password":"123123",
"column":["*"],
"splitPk":"ID",
"where":"ID<=?1888",
????????????????????????"connection":?[
????????????????????????????{
????????????????????????????????"jdbcUrl":?[
????????????????????????????????????"jdbc//192.168.1.1:3306/course-study?useUnicode=true&characterEncoding=utf8"
????????????????????????????????],?
????????????????????????????????"table":?["t_member"]
????????????????????????????}
????????????????????????]
????????????????????}
????????????????},?
????????????????"writer":?{
????????????????????"name":?"mysqlwriter",?
????????????????????"parameter":?{
????????????????????????"column":?["*"],?
????????????????????????"connection":?[
????????????????????????????{
????????????????????????????????"jdbcUrl":?"jdbc//192.168.1.2:3306/course-study?useUnicode=true&characterEncoding=utf8",
????????????????????????????????"table":?["t_member"]
????????????????????????????}
????????????????????????],?
????????????????????????"password":?"123123",
????????????????????????"preSql":?[
????????????????????????????"truncate?t_member"
????????????????????????],?
????????????????????????"session":?[
????????????????????????????"set?session?sql_mode='ANSI'"
????????????????????????],?
????????????????????????"username":?"root",?
????????????????????????"writeMode":?"insert"
????????????????????}
????????????????}
????????????}
????????],?
????????"setting":?{
????????????"speed":?{
????????????????"channel":?"5"
????????????}
????????}
????}
}

需要注意的部分就是：where（條件篩選） 和 preSql（同步前，要做的事） 參數(shù)。

2）驗證：

[root@MySQL-1 ~]# python /usr/local/data/bin/data.py where.json

輸出：

2021-12-16 1738.534 [job-0] INFO  JobContainer - PerfTrace not enable!
2021-12-16 1738.534 [job-0] INFO  StandAloneJobContainerCommunicator - Total 1888 records, 49543 bytes | Speed 1.61KB/s, 62 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.002s |  All Task WaitReaderTime 100.570s | Percentage 100.00%
2021-12-16 1738.537 [job-0] INFO  JobContainer - 
任務(wù)啟動時刻                    : 2021-12-16 1706
任務(wù)結(jié)束時刻                    : 2021-12-16 1738
任務(wù)總計耗時                    :                 32s
任務(wù)平均流量                    :            1.61KB/s
記錄寫入速度                    :             62rec/s
讀出記錄總數(shù)                    :                1888
讀寫失敗總數(shù)                    :                   0

目標數(shù)據(jù)庫上查看：



3）基于上面數(shù)據(jù)，再次進行增量同步：

主要是 where 配置："where": "ID > 1888 AND ID <= 2888"# 通過條件篩選來進行增量同步

同時需要將我上面的 preSql 刪除(因為我上面做的操作時 truncate 表)

審核編輯：劉清