DataX

DataX是阿里開源的離線資料同步工具,可以實現包括 MySQL、Oracle、MongoDB、Hive、HDFS、HBase、Elasticsearch等各種異構資料來源之間的高效同步。

DataX原理

設計理念

為了解決異構資料來源同步問題，DataX將複雜的網狀同步鏈路變成星型鏈路，DataX作為中間傳輸載體負責連線各種資料來源。當需要接入一個新的資料來源的時候，只需將此資料來源對接到DataX，便能跟已有資料來源做到無縫資料同步。

框架設計

採用Framework + plugin架構構建。將資料來源讀取和寫入抽象成為Reader/Writer外掛，納入到整個同步框架中。

• Reader：資料採集模組，負責採集資料來源的資料，將資料傳送給Framework。
• Writer： 資料寫入模組，負責不斷從Framework取資料，並將資料寫入到目的端。
• Framework：Framework用於連線Reader和Writer，作為兩者的資料傳輸通道，並處理緩衝，流控，併發，資料轉換等核心技術問題。

核心架構

• 核心模組介紹
  
  1. DataX完成單個數據同步的作業，我們稱之為Job，DataX接受到一個Job之後，將啟動一個程序來完成整個作業同步過程。DataX Job模組是單個作業的中樞管理節點，承擔了資料清理、子任務切分(將單一作業計算轉化為多個子Task)、TaskGroup管理等功能。
  2. DataXJob啟動後，會根據不同的源端切分策略，將Job切分成多個小的Task(子任務)，以便於併發執行。Task便是DataX作業的最小單元，每一個Task都會負責一部分資料的同步工作。
  3. 切分多個Task之後，DataX Job會呼叫Scheduler模組，根據配置的併發資料量，將拆分成的Task重新組合，組裝成TaskGroup(任務組)。每一個TaskGroup負責以一定的併發執行完畢分配好的所有Task，預設單個任務組的併發數量為5。
  4. 每一個Task都由TaskGroup負責啟動，Task啟動後，會固定啟動Reader—>Channel—>Writer的執行緒來完成任務同步工作。
  5. DataX作業執行起來之後， Job監控並等待多個TaskGroup模組任務完成，等待所有TaskGroup任務完成後Job成功退出。否則，異常退出，程序退出值非0。
• 排程流程
  舉例來說，使用者提交了一個DataX作業，並且配置了20個併發，目的是將一個100張分表的mysql資料同步到odps裡面。 DataX的排程決策思路是：
  
  1. DataXJob根據分庫分表切分成了100個Task。
  2. 根據20個併發，DataX計算共需要分配4個TaskGroup。
  3. 4個TaskGroup平分切分好的100個Task，每一個TaskGroup負責以5個併發共計執行25個Task。

DataX 部署

工具部署

```shell
[[email protected] /data/software]# wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
[[email protected] /data/software]# tar -zxvf datax.tar.gz
[[email protected] /data/software/datax]# python bin/datax.py job/job.json #自檢
```

目錄結構

 [[email protected] /data/software/datax]# tree -L 3 -I '*jar*'
    .
    ├── bin 啟動指令碼
    │   ├── datax.py
    │   ├── dxprof.py
    │   └── perftrace.py
    ├── conf 核心配置
    │   ├── core.json
    │   └── logback.xml
    ├── job  job目錄
    │   └── job.json
    ├── lib 核心類庫
    ├── log
    ├── log_perf
    ├── plugin 插卡目錄
    │   ├── reader 
    │   │   ├── drdsreader
    │   │   ├── ftpreader
    │   │   ├── hbase094xreader
    │   │   ├── hbase11xreader
    │   │   ├── hdfsreader
    │   │   ├── mongodbreader
    │   │   ├── mysqlreader
    │   │   ├── odpsreader
    │   │   ├── oraclereader
    │   │   ├── ossreader
    │   │   ├── otsreader
    │   │   ├── otsstreamreader
    │   │   ├── postgresqlreader
    │   │   ├── rdbmsreader
    │   │   ├── sqlserverreader
    │   │   ├── streamreader
    │   │   └── txtfilereader
    │   └── writer
    │       ├── adswriter
    │       ├── drdswriter
    │       ├── ftpwriter
    │       ├── hbase094xwriter
    │       ├── hbase11xsqlwriter
    │       ├── hbase11xwriter
    │       ├── hdfswriter
    │       ├── mongodbwriter
    │       ├── mysqlwriter
    │       ├── ocswriter
    │       ├── odpswriter
    │       ├── oraclewriter
    │       ├── osswriter
    │       ├── otswriter
    │       ├── postgresqlwriter
    │       ├── rdbmswriter
    │       ├── sqlserverwriter
    │       ├── streamwriter
    │       └── txtfilewriter
    ├── script
    │   └── Readme.md
    └── tmp
        └── readme.txt

DataX全量同步 Mysql-HDFS

DataX配置

 {
    "job":{
        "setting":{
            "errorLimit":{
                "record":1,
                "percentage":0.2
            },
            "speed": {
                 "channel":1
            }
        },
        "content":[
            {
                "reader":{
                    "name":"mysqlreader",
                    "parameter":{
                        "username":"root",
                        "password":"111",
                        "column":["id","log_type","event_time","uid"],
            "where":"event_time>='2018-08-10 01:01:01' and event_time<='2018-08-10 01:10:01'",
            "connection":[
                            {
                                "table":[
                                    "log_0",
                                    "log_1",
                                    "log_2"
                                ],
                                "jdbcUrl":[
                                    "jdbc:mysql://localhost:3306/test"
                                ]
                            }
                        ]
                    }
                },
                "writer":{
                    "name":"hdfswriter",
                    "parameter":{
                        "defaultFS":"hdfs://node1:8020",
                        "fileType":"text",
                        "path":"/test/access_log",
                        "fileName":"log_",
                        "fieldDelimiter":"\t",
                        "writeMode": "append",
                        "column":[
                           {
                            "name": "id",
                            "type": "bigint"
                            },
                            {
                             "name": "log_type",
                             "type": "string"
                            },
                            {
                             "name": "event_time",
                             "type": "string"
                            },
                            {
                             "name": "uid",
                             "type": "string"
                            }
                        ]
                    }
                }
            }
        ]
    }
}

DataX執行結果

 [[email protected] /data/software/datax]# python bin/datax.py -j"-Xms125m -Xmx125m" job/mysql_hdfs.json

任務啟動時刻                    : 2018-08-10 23:51:58
任務結束時刻                    : 2018-08-10 23:52:24
任務總計耗時                    :                 25s
任務平均流量                    :                5B/s
記錄寫入速度                    :              0rec/s
讀出記錄總數                    :                   3
讀寫失敗總數                    :                   0

[[email protected] /data/software]# hdfs dfs -cat /test/access_log/*
1   appError    2018-08-10 01:01:01 2
2   appError    2018-08-10 01:10:01 2
3   appError    2018-08-10 01:01:01 2

DataX增量同步 Mysql-HDFS

思路

增量模板+shell+crontab。
1.定義datax任務執行模板檔案 如"where":"event_time>='${start_time}' and event_time<='${end_time}'"。
2.每次定時任務啟動時根據上次偏移量替換模板檔案中增量變數。
3.執行完後記錄此次偏移量。

注意

1.時間增量,需記錄每次時間偏移量。
2.自增id增量,需記錄每次自增id偏移量。
3.不論是時間增量還是自增id增量,記得給資料同步留下足夠的時間。如從庫同步延遲。
4.實際中,要結合資料庫索引情況合理設定增量條件，提高每次查詢速度。
5.若沒有任何增量規律,datax只能做全量同步。