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

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

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

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

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

MySQL 讀寫操作不要影響線上業(yè)務，不要觸發(fā) MySQL 的運維告警，不想天天被 DBA 噴

希望支持更多的數(shù)據(jù)源，如 HBase、ES、文本文件

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

統(tǒng)計信息采集，例如運行時間、數(shù)據(jù)量、消耗資源

臟數(shù)據(jù)校驗和上報

希望運行日志能接入公司的日志平臺，方便監(jiān)控

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

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

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

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

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

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

符合平臺規(guī)則的狀態(tài)上報，如啟動/運行中/結束，運行時需上報進度，結束需上報成功失敗

符合平臺規(guī)則的運行日志實時上報，用于展示

統(tǒng)計、校驗、流控等子模塊的參數(shù)可從平臺傳入，并需要對結果做持久化

需要對異常輸入做好兼容，例如 MySQL 主從切換、表結構變更

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

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

表結構、表命名規(guī)則、地址轉(zhuǎn)換這些運行時前置校驗邏輯，以及運行結果的持久化，放在元數(shù)據(jù)系統(tǒng)（參考《有贊數(shù)據(jù)倉庫元數(shù)據(jù)系統(tǒng)實踐》），而運行狀態(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)獲取相應的路徑

Hive 的表結構獲取，需要依賴元數(shù)據(jù)系統(tǒng)。還需對 Hive 表結構做校驗，后面會詳細說明

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

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

按 DataX 的設計理念，reader 和 writer 相互不用關心，但實際使用經(jīng)常需要關聯(lián)考慮才能避免運行出錯。MySQL 加減字段，或者字段類型變更，都會導致 MySQL 和 Hive 的表結構不一致，需要避免這種不一致的運行出錯。

非分區(qū)表都是全量導入，以 mysqlreader 配置為準。如果 MySQL 配置字段與 Hive 實際結構不一致，則把 Hive 表 drop 掉后重建。表重建可能帶來下游 Hive SQL 出錯的風險，這個靠 SQL 的定時檢查規(guī)避。

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

Hive 分區(qū)表不能隨意變更表結構，變更可能會導致舊分區(qū)數(shù)據(jù)讀取異常。所以寫Hive 分區(qū)表時，以 Hive 表結構為準，表結構不一致則直接報錯。我們采取了如下的策略

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

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

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

元數(shù)據(jù)維護了標準的 RDS 中間件地址

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

每次 DataX 任務啟動時，獲取最新的主庫和從庫地址

定期的 MySQL 連通性校驗

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

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

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

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