一、認(rèn)識DataX

• DataX 是阿里云 DataWorks數(shù)據(jù)集成 的開源版本，在阿里巴巴集團(tuán)內(nèi)被廣泛使用的離線數(shù)據(jù)同步工具/平臺。DataX 實(shí)現(xiàn)了包括 MySQL、Oracle、OceanBase、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、Hologres、DRDS, databend 等各種異構(gòu)數(shù)據(jù)源之間高效的數(shù)據(jù)同步功能。

特征：

• DataX本身作為數(shù)據(jù)同步框架，將不同數(shù)據(jù)源的同步抽象為從源頭數(shù)據(jù)源讀取數(shù)據(jù)的Reader插件，以及向目標(biāo)端寫入數(shù)據(jù)的Writer插件，理論上DataX框架可以支持任意數(shù)據(jù)源類型的數(shù)據(jù)同步工作。同時(shí)DataX插件體系作為一套生態(tài)系統(tǒng), 每接入一套新數(shù)據(jù)源該新加入的數(shù)據(jù)源即可實(shí)現(xiàn)和現(xiàn)有的數(shù)據(jù)源互通。

二、DataX3概覽

DataX 是一個(gè)異構(gòu)數(shù)據(jù)源離線同步工具，致力于實(shí)現(xiàn)包括關(guān)系型數(shù)據(jù)庫(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各種異構(gòu)數(shù)據(jù)源之間穩(wěn)定高效的數(shù)據(jù)同步功能。

設(shè)計(jì)理念：

• 為了解決異構(gòu)數(shù)據(jù)源同步問題，DataX將復(fù)雜的網(wǎng)狀的同步鏈路變成了星型數(shù)據(jù)鏈路，DataX作為中間傳輸載體負(fù)責(zé)連接各種數(shù)據(jù)源。當(dāng)需要接入一個(gè)新的數(shù)據(jù)源的時(shí)候，只需要將此數(shù)據(jù)源對接到DataX，便能跟已有的數(shù)據(jù)源做到無縫數(shù)據(jù)同步。

當(dāng)前使用現(xiàn)狀：

• DataX在阿里巴巴集團(tuán)內(nèi)被廣泛使用，承擔(dān)了所有大數(shù)據(jù)的離線同步業(yè)務(wù)，并已持續(xù)穩(wěn)定運(yùn)行了6年之久。目前每天完成同步8w多道作業(yè)，每日傳輸數(shù)據(jù)量超過300TB。

三、DataX3框架設(shè)計(jì)

DataX本身作為離線數(shù)據(jù)同步框架，采用Framework + plugin架構(gòu)構(gòu)建。將數(shù)據(jù)源讀取和寫入抽象成為Reader/Writer插件，納入到整個(gè)同步框架中。

• Reader：Reader作為數(shù)據(jù)采集模塊，負(fù)責(zé)采集數(shù)據(jù)源的數(shù)據(jù)，將數(shù)據(jù)發(fā)送給Framework。
• Writer： Writer為數(shù)據(jù)寫入模塊，負(fù)責(zé)不斷向Framework取數(shù)據(jù)，并將數(shù)據(jù)寫入到目的端。
• Framework：Framework用于連接reader和writer，作為兩者的數(shù)據(jù)傳輸通道，并處理緩沖，流控，并發(fā)，數(shù)據(jù)轉(zhuǎn)換等核心技術(shù)問題。

四、DataX3插件體系

經(jīng)過幾年積累，DataX目前已經(jīng)有了比較全面的插件體系，主流的RDBMS數(shù)據(jù)庫、NOSQL、大數(shù)據(jù)計(jì)算系統(tǒng)都已經(jīng)接入。DataX目前支持?jǐn)?shù)據(jù)如下：

五、DataX3核心架構(gòu)

DataX 3開源版本支持單機(jī)多線程模式完成同步作業(yè)運(yùn)行，本小節(jié)按一個(gè)DataX作業(yè)生命周期的時(shí)序圖，從整體架構(gòu)設(shè)計(jì)非常簡要說明DataX各個(gè)模塊相互關(guān)系。

核心模塊介紹：

• DataX完成單個(gè)數(shù)據(jù)同步的作業(yè)，我們稱之為Job，DataX接受到一個(gè)Job之后，將啟動一個(gè)進(jìn)程來完成整個(gè)作業(yè)同步過程。DataX Job模塊是單個(gè)作業(yè)的中樞管理節(jié)點(diǎn)，承擔(dān)了數(shù)據(jù)清理、子任務(wù)切分(將單一作業(yè)計(jì)算轉(zhuǎn)化為多個(gè)子Task)、TaskGroup管理等功能。
• DataXJob啟動后，會根據(jù)不同的源端切分策略，將Job切分成多個(gè)小的Task(子任務(wù))，以便于并發(fā)執(zhí)行。Task便是DataX作業(yè)的最小單元，每一個(gè)Task都會負(fù)責(zé)一部分?jǐn)?shù)據(jù)的同步工作。
• 切分多個(gè)Task之后，DataX Job會調(diào)用Scheduler模塊，根據(jù)配置的并發(fā)數(shù)據(jù)量，將拆分成的Task重新組合，組裝成TaskGroup(任務(wù)組)。每一個(gè)TaskGroup負(fù)責(zé)以一定的并發(fā)運(yùn)行完畢分配好的所有Task，默認(rèn)單個(gè)任務(wù)組的并發(fā)數(shù)量為5。
• 每一個(gè)Task都由TaskGroup負(fù)責(zé)啟動，Task啟動后，會固定啟動Reader—>Channel—>Writer的線程來完成任務(wù)同步工作。
• DataX作業(yè)運(yùn)行起來之后， Job監(jiān)控并等待多個(gè)TaskGroup模塊任務(wù)完成，等待所有TaskGroup任務(wù)完成后Job成功退出。否則，異常退出，進(jìn)程退出值非0

DataX調(diào)度流程：

舉例來說，用戶提交了一個(gè)DataX作業(yè)，并且配置了20個(gè)并發(fā)，目的是將一個(gè)100張分表的mysql數(shù)據(jù)同步到odps里面。 DataX的調(diào)度決策思路是：

• DataXJob根據(jù)分庫分表切分成了100個(gè)Task。
• 根據(jù)20個(gè)并發(fā)，DataX計(jì)算共需要分配4個(gè)TaskGroup。
• 4個(gè)TaskGroup平分切分好的100個(gè)Task，每一個(gè)TaskGroup負(fù)責(zé)以5個(gè)并發(fā)共計(jì)運(yùn)行25個(gè)Task。

六、DataX 3六大核心優(yōu)勢

1.可靠的數(shù)據(jù)質(zhì)量監(jiān)控

完美解決數(shù)據(jù)傳輸個(gè)別類型失真問題

• DataX舊版對于部分?jǐn)?shù)據(jù)類型(比如時(shí)間戳)傳輸一直存在毫秒階段等數(shù)據(jù)失真情況，新版本DataX3已經(jīng)做到支持所有的強(qiáng)數(shù)據(jù)類型，每一種插件都有自己的數(shù)據(jù)類型轉(zhuǎn)換策略，讓數(shù)據(jù)可以完整無損的傳輸?shù)侥康亩恕?/li>

提供作業(yè)全鏈路的流量、數(shù)據(jù)量的運(yùn)行時(shí)監(jiān)控

• DataX3運(yùn)行過程中可以將作業(yè)本身狀態(tài)、數(shù)據(jù)流量、數(shù)據(jù)速度、執(zhí)行進(jìn)度等信息進(jìn)行全面的展示，讓用戶可以實(shí)時(shí)了解作業(yè)狀態(tài)。并可在作業(yè)執(zhí)行過程中智能判斷源端和目的端的速度對比情況，給予用戶更多性能排查信息。

提供臟數(shù)據(jù)探測：

• 在大量數(shù)據(jù)的傳輸過程中，必定會由于各種原因?qū)е潞芏鄶?shù)據(jù)傳輸報(bào)錯(cuò)(比如類型轉(zhuǎn)換錯(cuò)誤)，這種數(shù)據(jù)DataX認(rèn)為就是臟數(shù)據(jù)。DataX目前可以實(shí)現(xiàn)臟數(shù)據(jù)精確過濾、識別、采集、展示，為用戶提供多種的臟數(shù)據(jù)處理模式，讓用戶準(zhǔn)確把控?cái)?shù)據(jù)質(zhì)量大關(guān)！

2.豐富的數(shù)據(jù)轉(zhuǎn)換功能

DataX作為一個(gè)服務(wù)于大數(shù)據(jù)的ETL工具，除了提供數(shù)據(jù)快照搬遷功能之外，還提供了豐富數(shù)據(jù)轉(zhuǎn)換的功能，讓數(shù)據(jù)在傳輸過程中可以輕松完成數(shù)據(jù)脫敏，補(bǔ)全，過濾等數(shù)據(jù)轉(zhuǎn)換功能，另外還提供了自動groovy函數(shù)，讓用戶自定義轉(zhuǎn)換函數(shù)。

3.精準(zhǔn)的速度控制

還在為同步過程對在線存儲壓力影響而擔(dān)心嗎？新版本DataX3提供了包括通道(并發(fā))、記錄流、字節(jié)流三種流控模式，可以隨意控制你的作業(yè)速度，讓你的作業(yè)在庫可以承受的范圍內(nèi)達(dá)到最佳的同步速度。

"speed": {"channel": 5,"byte": 1048576,"record": 10000
}

4.強(qiáng)勁的同步性能

DataX3每一種讀插件都有一種或多種切分策略，都能將作業(yè)合理切分成多個(gè)Task并行執(zhí)行，單機(jī)多線程執(zhí)行模型可以讓DataX速度隨并發(fā)成線性增長。在源端和目的端性能都足夠的情況下，單個(gè)作業(yè)一定可以打滿網(wǎng)卡。另外，DataX團(tuán)隊(duì)對所有的已經(jīng)接入的插件都做了極致的性能優(yōu)化，并且做了完整的性能測試。

5.健壯的容錯(cuò)機(jī)制

DataX作業(yè)是極易受外部因素的干擾，網(wǎng)絡(luò)閃斷、數(shù)據(jù)源不穩(wěn)定等因素很容易讓同步到一半的作業(yè)報(bào)錯(cuò)停止。因此穩(wěn)定性是DataX的基本要求，在DataX3的設(shè)計(jì)中，重點(diǎn)完善了框架和插件的穩(wěn)定性。目前DataX3可以做到線程級別、進(jìn)程級別(暫時(shí)未開放)、作業(yè)級別多層次局部/全局的重試，保證用戶的作業(yè)穩(wěn)定運(yùn)行。

• 線程內(nèi)部重試：DataX的核心插件都經(jīng)過團(tuán)隊(duì)的全盤review，不同的網(wǎng)絡(luò)交互方式都有不同的重試策略。
• 線程級別重試：目前DataX已經(jīng)可以實(shí)現(xiàn)TaskFailover，針對于中間失敗的Task，DataX框架可以做到整個(gè)Task級別的重新調(diào)度。

6.極簡的使用體驗(yàn)

• 易用：下載即可用，支持linux和windows，只需要短短幾步驟就可以完成數(shù)據(jù)的傳輸。

• 詳細(xì)：DataX在運(yùn)行日志中打印了大量信息，其中包括傳輸速度，Reader、Writer性能，進(jìn)程CPU，JVM和GC情況等等。

  • 傳輸過程中打印傳輸速度、進(jìn)度等

• 傳輸過程中會打印進(jìn)程相關(guān)的CPU、JVM等

• 在任務(wù)結(jié)束之后，打印總體運(yùn)行情況
  

七、DataX3同步Mysql數(shù)據(jù)庫數(shù)據(jù)到Mysql數(shù)據(jù)庫

配置文件

{"job": {"setting": {"speed": {"channel": 6},"errorLimit": {"record": 0,"percentage": 0}},"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "aa","password": "1","splitPk": "id","column": ["id","user_id","tag_id","question_id","direction","is_in_tags","created_at","updated_at"],"connection": [{"table": ["`user_question`"],"jdbcUrl": ["jdbc:mysql://mysql-01-dev.com:3306/optics_dev?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai"]}]}},"writer": {"parameter": {"writeMode": "insert","username": "debug_test_rw","password": "1","column": ["id","user_id","tag_id","question_id","direction","is_in_tags","created_at","updated_at"],"preSql": ["delete from user_question_bak"],"connection": [{"table": ["user_question_bak"],"jdbcUrl": "jdbc:mysql://dd-mysql-01-test.com:3306/optics_test?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai"}]},"name": "mysqlwriter"}}]}
}

八、DataX3同步Mysql數(shù)據(jù)庫數(shù)據(jù)到Starrocks數(shù)據(jù)庫

{"job": {"setting": {"speed": {"channel": 8},"errorLimit": {"record": 0,"percentage": 0}},"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "debug_test_rw","password": "11111","splitPk": "id","column": ["id","tag_id","user_id","app_user_id","is_owner","sort","title","created_at","updated_at"],"connection": [{"table": ["`tag_user_20230116`"],"jdbcUrl": ["jdbc:mysql://mysql-01-test.com:3306/optics_test?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai"]}]}},"writer": {"name": "starrockswriter","parameter": {"username": "dd_scr_starrocks_rw","password": "11111","database": "dd_scr_starrocks","table": "tag_user","column": ["id","tag_id","user_id","app_user_id","is_owner","sort","title","created_at","updated_at"],"preSql": ["truncate table dd_scr_starrocks.tag_user"],"postSql": [],"jdbcUrl": "jdbc:mysql://10.129.66.144:9030/","loadUrl": ["10.129.66.144:8030", "10.129.88.67:8030", "10.129.68.124:8030"],"loadProps": {"format": "json","strip_outer_array": true}}}}]}
}