摘要：與大數(shù)據(jù)體系交互上報(bào)運(yùn)行統(tǒng)計(jì)數(shù)據(jù)自帶了運(yùn)行結(jié)果的統(tǒng)計(jì)數(shù)據(jù)，我們希望把這些統(tǒng)計(jì)數(shù)據(jù)上報(bào)到元數(shù)據(jù)系統(tǒng)，作為的過程元數(shù)據(jù)存儲(chǔ)下來?；谖覀兊拈_發(fā)策略，不要把有贊元數(shù)據(jù)系統(tǒng)的嵌入源碼，而是在之外獲取，截取出打印的統(tǒng)計(jì)信息再上報(bào)。

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

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

多次因 MySQL 變更引起的數(shù)據(jù)同步異常。MySQL 需要支持讀寫分離與分表分庫模式，而且要兼容可能的數(shù)據(jù)庫遷移、節(jié)點(diǎn)宕機(jī)以及主從切換

有不少異常是由表結(jié)構(gòu)變更導(dǎo)致。MySQL 或 Hive 的表結(jié)構(gòu)都可能發(fā)生變更，需要兼容多數(shù)的表結(jié)構(gòu)不一致情況

MySQL 讀寫操作不要影響線上業(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ì)劃基于開源做改造，考察的對(duì)象主要是 DataX 和 Sqoop，它們之間的功能對(duì)比如下

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

使用 DataX 最重要的是解決分布式部署和運(yùn)行問題，DataX 本身是單進(jìn)程的客戶端運(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ù)開發(fā)。關(guān)于調(diào)度系統(tǒng)，可參考文章《大數(shù)據(jù)開發(fā)平臺(tái)(Data Platform)在有贊的最佳踐》

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

為了與已有的數(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)變更

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

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

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

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

根據(jù)表名，拼接出 HDFS 路徑。有贊的數(shù)據(jù)倉庫規(guī)范里有一條，禁止使用外部表，這使得 HDFS 路徑拼接變得容易。若是外部表，就需要從元數(shù)據(jù)系統(tǒng)獲取相應(yīng)的路徑

Hive 的表結(jié)構(gòu)獲取，需要依賴元數(shù)據(jù)系統(tǒng)。還需對(duì) Hive 表結(jié)構(gòu)做校驗(yàn)，后面會(huì)詳細(xì)說明

寫 Hive 的配置里不會(huì)指定 Hive 的文件格式、分隔符，需要讀取元數(shù)據(jù)，獲取這些信息填入 HDFS 的寫配置文件

支持新建不存在的 Hive 表或分區(qū)，能構(gòu)建出符合數(shù)據(jù)倉庫規(guī)范的建表語句

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

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

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

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

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

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

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

元數(shù)據(jù)維護(hù)了標(biāo)準(zhǔn)的 RDS 中間件地址

主庫、從庫、RDS 中間件三者地址可以關(guān)聯(lián)和任意轉(zhuǎn)換

每次 DataX 任務(wù)啟動(dòng)時(shí)，獲取最新的主庫和從庫地址

定期的 MySQL 連通性校驗(yàn)

與 DBA 建立協(xié)作關(guān)系，變更提前通知

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

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

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