ChunJun 是?款穩(wěn)定、易?、?效、批流?體的數(shù)據(jù)集成框架，基于計算引擎 Flink 實現(xiàn)多種異構(gòu)數(shù)據(jù)源之間的數(shù)據(jù)同步與計算。ChunJun 可以把不同來源、格式、特點性質(zhì)的數(shù)據(jù)在邏輯上或物理上有機地集中，從?為企業(yè)提供全?的數(shù)據(jù)共享，目前已在上千家公司部署且穩(wěn)定運?。

在之前，我們曾經(jīng)為大家介紹過如何利用 ChunJun 實現(xiàn)數(shù)據(jù)實時同步（點擊看正文），本篇將為大家介紹姊妹篇，如何利? ChunJun 實現(xiàn)數(shù)據(jù)的離線同步。

ChunJun 離線同步案例

離線同步是 ChunJun 的?個重要特性，下?以最通?的 mysql -> hive 的同步任務(wù)來介紹離線同步。

配置環(huán)境

找?個空?錄，接下來要配置 Flink 和 ChunJun 的環(huán)境，下?以 /root/chunjun_demo/ 為例?。

● 配置 Flink

● 配置 ChunJun

解壓好的 ChunJun 有如下?錄： bin chunjun-dist chunjun-examples lib

● 配置環(huán)境變量

● 在 Yarn 上?啟動 Flink Session

輸出如下：

下?提交任務(wù)會?到 Flink Session 這個 Yarn Application Id (application_1683599622970_0270)。

● 其他配置

如果? parquet 格式，需要把 flink-parquet_2.12-1.12.7.jar 放?到 flink/lib 下?, 在上?的例?中，需要放到 $FLINK_HOME/lib ??。

提交任務(wù)

● 在 MySQL 準備數(shù)據(jù)

如果沒有 MySQL 的話，可以? docker 快速創(chuàng)建?個。

● 創(chuàng)建 Hive 表

● 在當前?錄( /root/chunjun_demo/ ) 配置?個任務(wù) mysql_hdfs.json

vim mysql_hdfs.json 輸?如下內(nèi)容：

因為我們要將 MySQL 同步到 Hive ??，但是如果直接同步 Hive 的話，內(nèi)部會? jdbc，? jdbc 的效率不?，因此我們可以直接把數(shù)據(jù)同步到 Hive 底層的 HDFS 上?，所以 writer ?到了 hdfswriter。腳本解析如下：

● 提交任務(wù)

● 查看任務(wù)

任務(wù)同步完成, 可以看?下 HDFS 上?的數(shù)據(jù)。

查看?下 Hive 表的數(shù)據(jù)。

注意, 如果是分區(qū)的 Hive 表，需要?動刷新?下 Hive 的元數(shù)據(jù), 使? MSCK 命令。（MSCK 是 Hive 中的?個命令，?于檢查表中的分區(qū)，并將其添加到 Hive 元數(shù)據(jù)中）

ChunJun 離線同步原理解析

HDFS 文件同步原理

· 對于?件系統(tǒng)，同步的時候會先把?件寫?到 path + [filename] ?錄??的 .data 的?件??，如果任務(wù)失敗，那么 .data ??的?件不會?效。

· 在 TaskManager 上?所有 task 任務(wù)結(jié)束的時候，會在 JobManager 執(zhí)? FinalizeOnMaster 的 finalizeGlobal ?法, 最終會調(diào)?到 moveAllTmpDataFileToDir , 把 .data ??的?件移除到 .data 的上?層。

增量同步

增量同步主要針對某些只有 Insert 操作的表，隨著業(yè)務(wù)增?，表內(nèi)數(shù)據(jù)越來越多。如果每次都同步整表的話，消耗的時間和資源會?較多。因此需要?個增量同步的功能，每次只讀取增加部分的數(shù)據(jù)。

● 實現(xiàn)原理

其實現(xiàn)原理實際上就是配合增量鍵在查詢的 sql 語句中拼接過濾條件，?如 where id > ? ，將之前已經(jīng)讀取過的數(shù)據(jù)過濾出去。

增量同步是針對于兩個及以上的同步作業(yè)來說的。對于初次執(zhí)?增量同步的作業(yè)??，實際上是整表同步，不同于其他作業(yè)的在于增量同步作業(yè)會在作業(yè)執(zhí)?完成后記錄?個 endLocation 指標，并將這個指標上傳到 prometheus 以供后續(xù)使?。

除第?次作業(yè)外，后續(xù)的所有增量同步作業(yè)都會取上?次作業(yè)的 endLocation 做為本次作業(yè)的過濾依據(jù)（startLocation）。?如第?次作業(yè)執(zhí)?完后，endLocation 為10，那么下?個作業(yè)就會構(gòu)建出例如 SELECT id,name,age from table where id > 10 的 SQL 語句，達到增量讀取的?的。

● 使用限制

· 只有 RDB 的 Reader 插件可以使?

· 通過構(gòu)建SQL過濾語句實現(xiàn)，因此只能?于RDB插件

· 增量同步只關(guān)?讀，不關(guān)?寫，因此只與Reader插件有關(guān)

· 增量字段只能為數(shù)值類型和時間類型

· 指標需要上傳到 prometheus，? prometheus 不?持字符串類型，因此只?持數(shù)據(jù)類型和時間類型，時間類型會轉(zhuǎn)換成時間戳后上傳

· 增量鍵的值可以重復，但必須遞增

· 由于使? '>' 的緣故，要求字段必須遞增

斷點續(xù)傳

斷點續(xù)傳是為了在離線同步的時候，針對?時間同步任務(wù)如超過1天，如果在同步過程中由于某些原因?qū)е氯蝿?wù)失敗，從頭再來的話成本?常?，因此需要?個斷點續(xù)傳的功能從任務(wù)失敗的地?繼續(xù)。

● 實現(xiàn)原理

· 基于 Flink 的 checkpoint，在 checkpoint 的時候 會存儲 source 端最后?條數(shù)據(jù)的某個字段值，sink 端插件執(zhí)?事務(wù)提交。

· 在任務(wù)失敗，后續(xù)通過 checkpoint 重新運?時，source 端在?成 select 語句的時候?qū)?state ?的值作為條件拼接進?數(shù)據(jù)的過濾，達到從上次失敗位點進?恢復。

· jdbcInputFormat 在拼接讀取 SQL 時，如果從 checkpoint 恢復的 state 不為空且 restoreColumn 不為空，則此時會將 checkpoint ?的 state 作為起點開始讀取數(shù)據(jù)。

● 適用場景

通過上述原理我們可以知道 source 端必須是 RDB 類型插件，因為是通過 select 語句拼接 where 條件進?數(shù)據(jù)過濾達到斷點續(xù)傳的，同時斷點續(xù)傳需要指定?個字段作為過濾條件，且此字段要求是遞增的。

· 任務(wù)需要開啟 checkpoint

· reader 為 RDB 的插件均?持且 writer ?持事務(wù)的插件(如 rdb filesystem 等)，如果下游是冪等性則 writer 插件也不需要?持事務(wù)

· 作為斷點續(xù)傳的字段在源表?的數(shù)據(jù)是遞增的，因為過濾條件是 >

《數(shù)據(jù)治理行業(yè)實踐白皮書》下載地址：https://fs80.cn/l134d5?

《數(shù)棧V6.0產(chǎn)品白皮書》下載地址：https://fs80.cn/cw0iw1

想了解或咨詢更多有關(guān)袋鼠云大數(shù)據(jù)產(chǎn)品、行業(yè)解決方案、客戶案例的朋友，瀏覽袋鼠云官網(wǎng)：https://www.dtstack.com/?src=szbzhan

同時，歡迎對大數(shù)據(jù)開源項目有興趣的同學加入「袋鼠云開源框架釘釘技術(shù) qun」，交流最新開源技術(shù)信息，qun 號碼：30537511，項目地址：https://github.com/DTStack