圖解 DataX 核心設(shè)計原理

DataX 是阿里巴巴開源的一個異構(gòu)數(shù)據(jù)源離線同步工具站故，致力于實現(xiàn)包括關(guān)系型數(shù)據(jù)庫（MySQL液荸、Oracle 等）关霸、HDFS忙干、Hive帕膜、ODPS霎烙、HBase遂填、FTP 等各種異構(gòu)數(shù)據(jù)源之間穩(wěn)定高效的數(shù)據(jù)同步功能尚猿。

前段時間我在 K8s 相關(guān)文章中有提到過數(shù)據(jù)同步的項目智绸，該項目就是基于 DataX 內(nèi)核構(gòu)建的野揪，由于公司數(shù)據(jù)同步的需求访忿，還需要在 DataX 原有的基礎(chǔ)上支持增量同步功能，同時支持分布式調(diào)度斯稳，在「使用 K8s 進(jìn)行作業(yè)調(diào)度實戰(zhàn)分享」這篇文章中已經(jīng)詳細(xì)描述其中的實現(xiàn)海铆。

基于我在項目中對 DataX 的實踐過程，給大家分享我所理解的 DataX 核心設(shè)計原理挣惰。

設(shè)計理念

異構(gòu)數(shù)據(jù)源離線同步是將源端數(shù)據(jù)同步到目的端卧斟，但是端與端的數(shù)據(jù)源類型種類繁多，在沒有 DataX 之前憎茂，端與端的鏈路將組成一個復(fù)雜的網(wǎng)狀結(jié)構(gòu)珍语，非常零散無法將同步核心邏輯抽象出來，DataX 的理念就是作為一個同步核心載體連接連接各類數(shù)據(jù)源竖幔，當(dāng)我們需要數(shù)據(jù)同步時板乙，只需要以插件的形式接入到 DataX 即可，將復(fù)雜的網(wǎng)狀結(jié)構(gòu)鏈路變成了一個星型結(jié)構(gòu)拳氢，如下圖所示：

image

架構(gòu)設(shè)計

用過 IDEA 的小伙都知道募逞，IDEA 有很多非常棒的插件，用戶可根據(jù)自身編程需求饿幅，下載相關(guān)的插件，DataX 也是使用這種可插拔的設(shè)計戒职，采用了 Framework + Plugin 的架構(gòu)設(shè)計栗恩，如下圖所示：

image

有了插件，DataX 可支持任意數(shù)據(jù)源到數(shù)據(jù)源洪燥，只要實現(xiàn)了 Reader/Writer Plugin磕秤，官方已經(jīng)實現(xiàn)了主流的數(shù)據(jù)源插件，比如 MySQL捧韵、Oracle市咆、SQLServer 等，當(dāng)然我們也可以開發(fā)一個 DataX 插件再来。

核心概念

DataX 核心主要由 Job蒙兰、Task Group、Task芒篷、Channel 等概念組成：

1搜变、Job

在 DataX 中用來描述一個源端到一個目的端的同步作業(yè)，是 DataX 數(shù)據(jù)同步面向用戶的最小業(yè)務(wù)單元针炉。一個Job 對應(yīng) 一個 JobContainer挠他， JobContainer 負(fù)責(zé) Job 的全局切分、調(diào)度篡帕、前置語句和后置語句等工作殖侵。

2贸呢、Task Group

一組 Task 的集合，根據(jù) DataX 的公平分配策略拢军，公平地分配 Task 到對應(yīng)的 TaskGroup 中楞陷。一個 TaskGroup 對應(yīng)一個 TaskGroupContainer，負(fù)責(zé)執(zhí)行一組 Task朴沿。

3猜谚、Task

Job 的最小執(zhí)行單元，一個 Job 可根據(jù) Reader 端切分策略赌渣，且分成若干個 Task魏铅，以便于并發(fā)執(zhí)行。

Job坚芜、Task Group览芳、Task 三者之間的關(guān)系可以用如下圖表示：

image

根據(jù)切分策略將一個 Job 切分成多個 Task，根據(jù)分配策略將多個 Task 組成一個 TaskGroup鸿竖。

4沧竟、Channel

DataX 會單獨啟動一條線程運行運行一個 Task，而 Task 會持有一個 Channel缚忧，用作 Reader 與 Writer 的數(shù)據(jù)傳輸媒介悟泵，DataX 的數(shù)據(jù)流向都是按照 Reader—>Channel—>Writer 的方向流轉(zhuǎn)，用如下圖表示：

image

Channel 作為傳輸通道闪水，即能充當(dāng)緩沖層糕非，同時還能對數(shù)據(jù)傳輸進(jìn)行限流操作。

5球榆、Transformer

DataX 的 transformer 模式同時還提供了強(qiáng)大的數(shù)據(jù)轉(zhuǎn)換功能朽肥，DataX 默認(rèn)提供了豐富的數(shù)據(jù)轉(zhuǎn)換實現(xiàn)類，用戶還可以根據(jù)項目自身需求持钉，擴(kuò)展數(shù)據(jù)轉(zhuǎn)換衡招。

image

調(diào)度流程

DataX 將用戶的 job.json 同步作業(yè)配置解析成一個 Job，DataX 通過 JobContainer 完成全局切分每强、調(diào)度始腾、前置語句和后置語句等工作，整體調(diào)度流程用如下圖表示：

image

1空执、切分策略

1）計算并發(fā)量（即 needChannelNumber 大芯阶隆）

DataX有流控模式，其中脆烟，可以設(shè)置 bps 限速山林，tps 限速：

bps 限速：needChannelNumber = 總 byteLimit / 單個 Channel byteLimit
tps 限速：needChannelNumber = 總 recordLimit / 單個 Channel recordLimit

如果以上都沒有設(shè)置，則會根據(jù)用戶在 job.setting.speed.channel 配置的并發(fā)數(shù)量設(shè)置 needChannelNumber。

2）根據(jù) needChannelNumber 將 Job 切分成多個 Task

這個步驟的具體切分邏輯交由相關(guān)插件去完成驼抹，例如 Rdb 對數(shù)據(jù)的拆分主要分成兩類：

如果用戶配置了具體的 Table 數(shù)量桑孩，那么就按照 Table 為最小單元進(jìn)行拆分（即一個 Table 對應(yīng)一個 Task），并生成對應(yīng)的 querySql框冀；
如果用戶還配置了 splitPk流椒，則會根據(jù) splitPk 進(jìn)行切分，具體邏輯是根據(jù) splitPk 區(qū)間對 Table 進(jìn)行拆分明也，并生成對應(yīng)的 querySql宣虾。

2、公平分配策略

DataX 在執(zhí)行調(diào)度之前温数，會調(diào)用 JobAssignUtil#assignFairly方法對切分好的 Task 公平分配給每個 TaskGroup绣硝。

在分配之前，會計算 TaskGroup 的數(shù)量撑刺，具體公式：

int taskGroupNumber = (int) Math.ceil(1.0 * channelNumber / channelsPerTaskGroup);

channelNumber 即為在切分策略中根據(jù)用戶配置計算得到的 needChannelNumber 并發(fā)數(shù)量大小鹉胖，channelsPerTaskGroup 為每個 TaskGroup 需要的并發(fā)數(shù)量，默認(rèn)為 5够傍。

求出 TaskGroup 的數(shù)量之后甫菠，就會執(zhí)行公平分配策略，將 Task 平均分配個每個 TaskGroup冕屯，最后執(zhí)行調(diào)度寂诱，完成整個同步作業(yè)。舉個公平分配策略的例子：

假設(shè) A 庫有表 0安聘、1痰洒、2，B 庫上有表 3搞挣、4带迟，C 庫上有表 5音羞、6囱桨、7，如果此時有 4 個 TaskGroup嗅绰，則 assign 后的結(jié)果為：

taskGroup-0: 0,  4,
taskGroup-1: 3,  6,
taskGroup-2: 5,  2,
taskGroup-3: 1,  7

舉個例子來描述 Job舍肠、Task、Task Group 之間的關(guān)系：

用戶構(gòu)建了一個數(shù)據(jù)同步作業(yè)窘面，該作業(yè)的目的是將 MySql 的 100 張表同步到 Oracle 庫中翠语，假設(shè)此時用戶設(shè)置了 20 個并發(fā)（即 channelNumber=20）：

DataX 根據(jù)表的數(shù)量切分成 100 個 Task；
DataX 默認(rèn)給每個 TaskGroup 分配 5 個 Channel财边，因此 taskGroupNumber = channelNumber / channelsPerTaskGroup = 20 / 5 = 4肌括；
根據(jù) DataX 的公平分配策略，會將 100 個 Task 平均分配給每個 TaskGroup酣难，因此每個 TaskGroup 處理 taskNumber / taskGroupNumber = 100 / 4 = 25 個 Task谍夭。

以上的例子用如下圖表示：

image

由于一個 Channel 對應(yīng)一個線程執(zhí)行黑滴，因此 DataX 的線程模型可以用如下圖表示：

image

作者簡介

作者張乘輝，擅長消息中間件技能紧索，負(fù)責(zé)公司百萬 TPS 級別 Kafka 集群的維護(hù)袁辈，作者維護(hù)的公號「后端進(jìn)階」不定期分享 Kafka、RocketMQ 系列不講概念直接真刀真槍的實戰(zhàn)總結(jié)以及細(xì)節(jié)上的源碼分析珠漂；同時作者也是阿里開源分布式事務(wù)框架 Seata Contributor晚缩，因此也會分享關(guān)于 Seata 的相關(guān)知識；當(dāng)然公號也會分享 WEB 相關(guān)知識比如 Spring 全家桶等媳危。內(nèi)容不一定面面俱到荞彼，但一定讓你感受到作者對于技術(shù)的追求是認(rèn)真的！

公眾號：后端進(jìn)階

技術(shù)博客：https://objcoding.com/

GitHub：https://github.com/objcoding/