一、需求

有贊大數(shù)據(jù)技術(shù)應(yīng)用的早期，我們使用 Sqoop 作為數(shù)據(jù)同步工具，滿(mǎn)足了 MySQL 與 Hive 之間數(shù)據(jù)同步的日常開(kāi)發(fā)需求。

隨著公司業(yè)務(wù)發(fā)展，數(shù)據(jù)同步的場(chǎng)景越來(lái)越多，主要是 MySQL、Hive 與文本文件之間的數(shù)據(jù)同步，Sqoop 已經(jīng)不能完全滿(mǎn)足我們的需求。在2017年初，我們已經(jīng)無(wú)法忍受 Sqoop 給我們帶來(lái)的折磨，準(zhǔn)備改造我們的數(shù)據(jù)同步工具。當(dāng)時(shí)有這么些很最痛的需求：

• 多次因 MySQL 變更引起的數(shù)據(jù)同步異常。MySQL 需要支持讀寫(xiě)分離與分表分庫(kù)模式，而且要兼容可能的數(shù)據(jù)庫(kù)遷移、節(jié)點(diǎn)宕機(jī)以及主從切換
• 有不少異常是由表結(jié)構(gòu)變更導(dǎo)致。MySQL 或 Hive 的表結(jié)構(gòu)都可能發(fā)生變更，需要兼容多數(shù)的表結(jié)構(gòu)不一致情況
• MySQL 讀寫(xiě)操作不要影響線上業(yè)務(wù)，不要觸發(fā) MySQL 的運(yùn)維告警，不想天天被 DBA 噴
• 希望支持更多的數(shù)據(jù)源，如 HBase、ES、文本文件

作為數(shù)據(jù)平臺(tái)管理員，還希望收集到更多運(yùn)行細(xì)節(jié)，方便日常維護(hù)：

• 統(tǒng)計(jì)信息采集，例如運(yùn)行時(shí)間、數(shù)據(jù)量、消耗資源
• 臟數(shù)據(jù)校驗(yàn)和上報(bào)
• 希望運(yùn)行日志能接入公司的日志平臺(tái)，方便監(jiān)控

二、選型

基于上述的數(shù)據(jù)同步需求，我們計(jì)劃基于開(kāi)源做改造，考察的對(duì)象主要是 DataX 和 Sqoop，它們之間的功能對(duì)比如下

功能DataXSqoop

運(yùn)行模式	單進(jìn)程多線程	MapReduce
MySQL讀寫(xiě)	單機(jī)壓力大；讀寫(xiě)粒度容易控制	MapReduce 模式重，寫(xiě)出錯(cuò)處理麻煩
Hive讀寫(xiě)	單機(jī)壓力大	擴(kuò)展性好
文件格式	orc支持	orc不支持，可添加
分布式	不支持，可以通過(guò)調(diào)度系統(tǒng)規(guī)避	支持
流控	有流控功能	需要定制
統(tǒng)計(jì)信息	已有一些統(tǒng)計(jì)，上報(bào)需定制	沒(méi)有，分布式的數(shù)據(jù)收集不方便
數(shù)據(jù)校驗(yàn)	在core部分有校驗(yàn)功能	沒(méi)有，分布式的數(shù)據(jù)收集不方便
監(jiān)控	需要定制	需要定制
社區(qū)	開(kāi)源不久，社區(qū)不活躍	一直活躍，核心部分變動(dòng)很少

DataX 主要的缺點(diǎn)在于單機(jī)運(yùn)行，而這個(gè)可以通過(guò)調(diào)度系統(tǒng)規(guī)避，其他方面的功能均優(yōu)于 Sqoop，最終我們選擇了基于 DataX 開(kāi)發(fā)。

三、前期設(shè)計(jì)

3.1 運(yùn)行形態(tài)

使用 DataX 最重要的是解決分布式部署和運(yùn)行問(wèn)題，DataX 本身是單進(jìn)程的客戶(hù)端運(yùn)行模式，需要考慮如何觸發(fā)運(yùn)行 DataX。

我們決定復(fù)用已有的離線任務(wù)調(diào)度系統(tǒng)，任務(wù)觸發(fā)由調(diào)度系統(tǒng)負(fù)責(zé)，DataX 只負(fù)責(zé)數(shù)據(jù)同步。這樣就復(fù)用了系統(tǒng)能力，避免重復(fù)開(kāi)發(fā)。關(guān)于調(diào)度系統(tǒng)，可參考文章《大數(shù)據(jù)開(kāi)發(fā)平臺(tái)(Data Platform)在有贊的最佳踐》

在每個(gè)數(shù)據(jù)平臺(tái)的 worker 服務(wù)器，都會(huì)部署一個(gè) DataX 客戶(hù)端，運(yùn)行時(shí)可同時(shí)啟動(dòng)多個(gè)進(jìn)程，這些都由調(diào)度系統(tǒng)控制。

3.2 執(zhí)行器設(shè)計(jì)

為了與已有的數(shù)據(jù)平臺(tái)交互，需要做一些定制修改：

• 符合平臺(tái)規(guī)則的狀態(tài)上報(bào)，如啟動(dòng)/運(yùn)行中/結(jié)束，運(yùn)行時(shí)需上報(bào)進(jìn)度，結(jié)束需上報(bào)成功失敗
• 符合平臺(tái)規(guī)則的運(yùn)行日志實(shí)時(shí)上報(bào)，用于展示
• 統(tǒng)計(jì)、校驗(yàn)、流控等子模塊的參數(shù)可從平臺(tái)傳入，并需要對(duì)結(jié)果做持久化
• 需要對(duì)異常輸入做好兼容，例如 MySQL 主從切換、表結(jié)構(gòu)變更

3.3 開(kāi)發(fā)策略

大致的運(yùn)行流程是：前置配置文件轉(zhuǎn)換、表結(jié)構(gòu)校驗(yàn) -> （輸入 -> DataX 核心+業(yè)務(wù)無(wú)關(guān)的校驗(yàn) -> 輸出） -> 后置統(tǒng)計(jì)/持久化

盡量保證 DataX 專(zhuān)注于數(shù)據(jù)同步，盡量不隱含業(yè)務(wù)邏輯，把有贊特有的業(yè)務(wù)邏輯放到 DataX 之外，數(shù)據(jù)同步過(guò)程無(wú)法滿(mǎn)足的需求，才去修改源碼。

表結(jié)構(gòu)、表命名規(guī)則、地址轉(zhuǎn)換這些運(yùn)行時(shí)前置校驗(yàn)邏輯，以及運(yùn)行結(jié)果的持久化，放在元數(shù)據(jù)系統(tǒng)（參考《有贊數(shù)據(jù)倉(cāng)庫(kù)元數(shù)據(jù)系統(tǒng)實(shí)踐》），而運(yùn)行狀態(tài)的監(jiān)控放在調(diào)度系統(tǒng)。

四、源碼改造之路

4.1 支持 Hive 讀寫(xiě)

DataX 并沒(méi)有自帶 Hive 的 reader 和 writer，而只有 HDFS 的 reader 和writer。我們選擇在 DataX 之外封裝，把 Hive 讀寫(xiě)操作的配置文件，轉(zhuǎn)換為 HDFS 讀寫(xiě)的配置文件，另外輔助上 Hive DDL 操作。具體的，我們做了如下改造：

4.1.1 Hive 讀操作

• 根據(jù)表名，拼接出 HDFS 路徑。有贊的數(shù)據(jù)倉(cāng)庫(kù)規(guī)范里有一條，禁止使用外部表，這使得 HDFS 路徑拼接變得容易。若是外部表，就需要從元數(shù)據(jù)系統(tǒng)獲取相應(yīng)的路徑
• Hive 的表結(jié)構(gòu)獲取，需要依賴(lài)元數(shù)據(jù)系統(tǒng)。還需對(duì) Hive 表結(jié)構(gòu)做校驗(yàn)，后面會(huì)詳細(xì)說(shuō)明

4.1.2 Hive 寫(xiě)操作

• 寫(xiě) Hive 的配置里不會(huì)指定 Hive 的文件格式、分隔符，需要讀取元數(shù)據(jù)，獲取這些信息填入 HDFS 的寫(xiě)配置文件
• 支持新建不存在的 Hive 表或分區(qū)，能構(gòu)建出符合數(shù)據(jù)倉(cāng)庫(kù)規(guī)范的建表語(yǔ)句

4.2 MySQL -> Hive兼容性

按 DataX 的設(shè)計(jì)理念，reader 和 writer 相互不用關(guān)心，但實(shí)際使用經(jīng)常需要關(guān)聯(lián)考慮才能避免運(yùn)行出錯(cuò)。MySQL 加減字段，或者字段類(lèi)型變更，都會(huì)導(dǎo)致 MySQL 和 Hive 的表結(jié)構(gòu)不一致，需要避免這種不一致的運(yùn)行出錯(cuò)。

4.2.1 MySQL -> Hive 非分區(qū)表

非分區(qū)表都是全量導(dǎo)入，以 mysqlreader 配置為準(zhǔn)。如果 MySQL 配置字段與 Hive 實(shí)際結(jié)構(gòu)不一致，則把 Hive 表 drop 掉后重建。表重建可能帶來(lái)下游 Hive SQL 出錯(cuò)的風(fēng)險(xiǎn)，這個(gè)靠 SQL 的定時(shí)檢查規(guī)避。

Hive 表重建時(shí)，需要做 MySQL 字段轉(zhuǎn)換為 Hive 類(lèi)型，比如 MySQL 的 varchar 轉(zhuǎn)為 Hive 的 string。這里有坑，Hive 沒(méi)有無(wú)符號(hào)類(lèi)型，注意 MySQL 的 int unsigned 的取值范圍，需要向上轉(zhuǎn)型，轉(zhuǎn)為 Hive 的 bigint；同理，MySQL 的 bigint unsigned 也需要向上轉(zhuǎn)型，我們根據(jù)實(shí)際業(yè)務(wù)情況大膽轉(zhuǎn)為 bigint。而 Hive 的 string 是萬(wàn)能類(lèi)型，如果不知道怎么轉(zhuǎn)，用 string 是比較保險(xiǎn)的。

4.2.2 MySQL -> Hive 分區(qū)表

Hive 分區(qū)表不能隨意變更表結(jié)構(gòu)，變更可能會(huì)導(dǎo)致舊分區(qū)數(shù)據(jù)讀取異常。所以寫(xiě)Hive 分區(qū)表時(shí)，以 Hive 表結(jié)構(gòu)為準(zhǔn)，表結(jié)構(gòu)不一致則直接報(bào)錯(cuò)。我們采取了如下的策略

MySQL字段Hive實(shí)際字段處理方法

a,b	a,b	正常
a,b,c	a,b	忽略MySQL的多余字段，以Hive為準(zhǔn)
b,a	a,b	順序不對(duì)，調(diào)整
a	a,b	MySQL少一個(gè)，報(bào)錯(cuò)
a,c	a,b	不匹配, 報(bào)錯(cuò)
未指定字段	a,b	以Hive為準(zhǔn)

這么做偏保守，對(duì)于無(wú)害的Hive分區(qū)表變更，其實(shí)可以大膽去做，比如int類(lèi)型改bigint、orc表加字段。

4.3 適配 MySQL 集群

有贊并沒(méi)有獨(dú)立運(yùn)行的 MySQL 實(shí)例，都是由 RDS 中間件管理著 MySQL 集群，有讀寫(xiě)分離和分表分庫(kù)兩種模式。讀寫(xiě) MySQL 有兩種選擇，通過(guò) RDS 中間件讀寫(xiě)，以及直接讀寫(xiě) MySQL 實(shí)例。

方案優(yōu)先缺點(diǎn)

連實(shí)例	性能好；不影響線上業(yè)務(wù)	當(dāng)備庫(kù)維護(hù)或切換地址時(shí)，需要修改配置；開(kāi)發(fā)者不知道備庫(kù)地址
連 RDS	與普通應(yīng)用一致；屏蔽了后端維護(hù)	對(duì) RDS 造成額外壓力，有影響線上業(yè)務(wù)的風(fēng)險(xiǎn)；需要完全符合公司 SQL 規(guī)范

對(duì)于寫(xiě) MySQL，寫(xiě)入的數(shù)據(jù)量一般不大，DataX 選擇連 RDS，這樣就不用額外考慮主從復(fù)制。 對(duì)于讀 MySQL，考慮到有大量的全表同步任務(wù)，特別是凌晨離線任務(wù)高峰流量特別大，避免大流量對(duì) RDS 中間件的沖擊，DataX 選擇直連到 MySQL 實(shí)例去讀取數(shù)據(jù)。為了規(guī)避 MySQL 維護(hù)帶來(lái)的地址變更風(fēng)險(xiǎn)，我們又做了幾件事情：

• 元數(shù)據(jù)維護(hù)了標(biāo)準(zhǔn)的 RDS 中間件地址
• 主庫(kù)、從庫(kù)、RDS 中間件三者地址可以關(guān)聯(lián)和任意轉(zhuǎn)換
• 每次 DataX 任務(wù)啟動(dòng)時(shí)，獲取最新的主庫(kù)和從庫(kù)地址
• 定期的 MySQL 連通性校驗(yàn)
• 與 DBA 建立協(xié)作關(guān)系，變更提前通知

讀取 MySQL 時(shí)，對(duì)于讀寫(xiě)分離，每次獲取其中一個(gè)從庫(kù)地址并連接；對(duì)于分表分庫(kù)，我們有1024分片，就要轉(zhuǎn)換出1024個(gè)從庫(kù)地址，拼接出 DataX 的配置文件。

4.4 MySQL 運(yùn)維規(guī)范的兼容

4.4.1 避免慢 SQL

前提是有贊的 MySQL 建表規(guī)范，規(guī)定了建表必須有整型自增id主鍵。另一條運(yùn)維規(guī)范，SQL 運(yùn)行超過(guò)2s會(huì)被強(qiáng)行 kill 掉。

以讀取 MySQL 全表為例，我們把一條全表去取的 SQL，拆分為很多條小 SQL，而每條小 SQL 只走主鍵 id 的聚簇索引，代碼如下 select ... from table_name where id>? by id asc limit ?

4.4.2 避免過(guò)快讀寫(xiě)影響其他業(yè)務(wù)

執(zhí)行完一條 SQL 后會(huì)強(qiáng)制 sleep 一下，讓系統(tǒng)不能太忙。無(wú)論是 insert 的 batchSize，還是 select 每次分頁(yè)大小，我們都是動(dòng)態(tài)生成的，根據(jù)上一條運(yùn)行的時(shí)間，運(yùn)行太快就多 sleep，運(yùn)行太慢就少 sleep，同時(shí)調(diào)整下一個(gè)批次的數(shù)量。

這里還有改進(jìn)的空間，可以根據(jù)系統(tǒng)級(jí)的監(jiān)控指標(biāo)動(dòng)態(tài)調(diào)整速率，比如磁盤(pán)使用率、CPU 使用率、binlog 延遲等。實(shí)際運(yùn)行中，刪數(shù)據(jù)很容易引起 binlog 延遲，僅從 delete 語(yǔ)句運(yùn)行時(shí)間無(wú)法判斷是否刪的太快，具體原因尚未去深究。

4.5 更多的插件

除了最常用的 MySQL、Hive，以及邏輯比較簡(jiǎn)單的文本，我們還對(duì) HBase 的讀寫(xiě)根據(jù)業(yè)務(wù)情況做了簡(jiǎn)單改造。 我們還全新開(kāi)發(fā)了 eswriter，以及有贊 kvds 的 kvwriter，這些都是由相關(guān)存儲(chǔ)的開(kāi)發(fā)者負(fù)責(zé)開(kāi)發(fā)和維護(hù)插件。

4.6 與大數(shù)據(jù)體系交互

4.6.1 上報(bào)運(yùn)行統(tǒng)計(jì)數(shù)據(jù)

DataX 自帶了運(yùn)行結(jié)果的統(tǒng)計(jì)數(shù)據(jù)，我們希望把這些統(tǒng)計(jì)數(shù)據(jù)上報(bào)到元數(shù)據(jù)系統(tǒng)，作為 ETL 的過(guò)程元數(shù)據(jù)存儲(chǔ)下來(lái)。

基于我們的開(kāi)發(fā)策略，不要把有贊元數(shù)據(jù)系統(tǒng)的 api 嵌入 DataX 源碼，而是在 DataX 之外獲取 stdout，截取出打印的統(tǒng)計(jì)信息再上報(bào)。

4.6.2 與數(shù)據(jù)平臺(tái)的交互

數(shù)據(jù)平臺(tái)提供了 DataX 任務(wù)的編輯頁(yè)面，保存后會(huì)留下 DataX 運(yùn)行配置文件以及調(diào)度周期在平臺(tái)上。調(diào)度系統(tǒng)會(huì)根據(jù)調(diào)度周期和配置文件，定時(shí)啟動(dòng) DataX 任務(wù)，每個(gè) DataX 任務(wù)以獨(dú)立進(jìn)程的方式運(yùn)行，進(jìn)程退出后任務(wù)結(jié)束。運(yùn)行中，會(huì)把 DataX 的日志實(shí)時(shí)傳輸并展示到頁(yè)面上。

4.7 考慮更多異常

DataX 代碼中多數(shù)場(chǎng)景暴力的使用catch Exception，缺乏對(duì)各異常場(chǎng)景的兼容或重試，一個(gè)大任務(wù)執(zhí)行過(guò)程中出現(xiàn)網(wǎng)絡(luò)、IO等異常容易引起任務(wù)失敗。最常見(jiàn)的異常就是 SQLException，需要對(duì)異常做分類(lèi)處理，比如 SQL 異常考慮重試，批量處理異常改走單條依次處理，網(wǎng)絡(luò)異常考慮數(shù)據(jù)庫(kù)連接重建。 HDFS 對(duì)異常容忍度較高，DataX 較少捕獲異常。

4.8 測(cè)試場(chǎng)景改造

4.8.1 持續(xù)集成

為了發(fā)現(xiàn)低級(jí)問(wèn)題，例如表遷移了但任務(wù)還在、普通表改成了分區(qū)表，我們每天晚上20點(diǎn)以后，會(huì)把當(dāng)天運(yùn)行的所有重要 DataX 任務(wù)“重放”一遍。

這不是原樣重放，而是在配置文件里加入了一個(gè)測(cè)試的標(biāo)識(shí)，DataX 啟動(dòng)后，reader 部分只會(huì)讀取一行數(shù)據(jù)，而 writer 會(huì)把目標(biāo)地址指向一個(gè)測(cè)試的空間。這個(gè)測(cè)試能保證 DataX基本功能沒(méi)問(wèn)題，以及整個(gè)運(yùn)行環(huán)境沒(méi)有問(wèn)題。

4.8.2 全鏈路壓測(cè)場(chǎng)景

有贊全鏈路壓測(cè)系統(tǒng)通過(guò) Hive 來(lái)生成數(shù)據(jù)，通過(guò) DataX 把生成好的數(shù)據(jù)導(dǎo)入影子庫(kù)。影子庫(kù)是一種建在生產(chǎn) MySQL 里的 database，對(duì)普通應(yīng)用不可見(jiàn)，加上 SQL 的特殊 hint 才可以訪問(wèn)。

生產(chǎn)環(huán)境的全鏈路壓測(cè)是個(gè)高危操作，一旦配置文件有誤可能會(huì)破壞真實(shí)的生產(chǎn)數(shù)據(jù)。DataX 的 MySQL 讀寫(xiě)參數(shù)里，加上了全鏈路壓測(cè)的標(biāo)記時(shí)，只能讀寫(xiě)特定的 MySQL 和 Hive 庫(kù)，并配置數(shù)據(jù)平臺(tái)做好醒目的提醒。

五、線上運(yùn)行情況

DataX 是在2017年二季度正式上線，到2017年Q3完成了上述的大部分特性開(kāi)發(fā)，后續(xù)僅做了少量修補(bǔ)。到2019年Q1，已經(jīng)穩(wěn)定運(yùn)行了超過(guò)20個(gè)月時(shí)間，目前每天運(yùn)行超過(guò)6000個(gè) DataX 任務(wù)，傳輸了超過(guò)100億行數(shù)據(jù)，是數(shù)據(jù)平臺(tái)里比較穩(wěn)定的一個(gè)組件。

期間出現(xiàn)過(guò)一些小問(wèn)題，有一個(gè)印象深刻。原生的 hdfsreader 讀取超大 orc 文件有 bug，orc 的讀 api 會(huì)把大文件分片成多份，默認(rèn)大于256MB會(huì)分片，而 datax 僅讀取了第一個(gè)分片，修改為讀取所有分片解決問(wèn)題。因?yàn)?56MB足夠大，這個(gè)問(wèn)題很少出現(xiàn)很隱蔽。除此之外沒(méi)有發(fā)現(xiàn)大的 bug，平時(shí)遇到的問(wèn)題，多數(shù)是運(yùn)行環(huán)境或用戶(hù)理解的問(wèn)題，或是可以克服的小問(wèn)題。

六、下一步計(jì)劃

對(duì)于 DataX 其實(shí)并沒(méi)有再多的開(kāi)發(fā)計(jì)劃。在需求列表里積累了十幾條改進(jìn)需求，而這些需求即便1年不去改進(jìn)，也不會(huì)影響線上運(yùn)行，諸如臟數(shù)據(jù)可讀性改進(jìn)、支持 HDFS HA。這些不重要不緊急的需求，暫時(shí)不會(huì)再投入去做。

DataX 主要解決批量同步問(wèn)題，無(wú)法滿(mǎn)足多數(shù)增量同步和實(shí)時(shí)同步的需求。對(duì)于增量同步我們也有了成熟方案，會(huì)有另一篇文章介紹我們自研的增量同步產(chǎn)品。

總結(jié)

以上是生活随笔為你收集整理的DataX在有赞大数据平台的实践的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。