開源共建 | 中國移動馮江濤:ChunJun(原FlinkX)在數據入湖中的應用

来源:https://www.cnblogs.com/DTinsight/archive/2022/11/23/16917423.html
-Advertisement-
Play Games

ChunJun(原FlinkX)是一個基於 Flink 提供易用、穩定、高效的批流統一的數據集成工具。2018年4月,秉承著開源共用的理念,數棧技術團隊在github上開源了FlinkX,承蒙各位開發者的合作共建,FlinkX得到了快速發展。 兩年後的2022年4月,技術團隊決定對FlinkX進行整 ...


ChunJun(原FlinkX)是一個基於 Flink 提供易用、穩定、高效的批流統一的數據集成工具。2018年4月,秉承著開源共用的理念,數棧技術團隊在github上開源了FlinkX,承蒙各位開發者的合作共建,FlinkX得到了快速發展。

兩年後的2022年4月,技術團隊決定對FlinkX進行整體升級,並更名為ChunJun,希望繼續和各位優秀開發者合作,進一步推動數據集成/同步的技術發展。

因該文創作於於FlinkX更名為ChunJun之前,因此文中仍用FlinkX來進行分享,重要的事情說三遍:

FlinkX即是ChunJun

FlinkX即是ChunJun

FlinkX即是ChunJun

進入正文分享⬇️⬇️⬇️

file

分享嘉賓:馮江濤 中國移動雲能力中心

編輯整理:陳凱翔 亞廈股份

出品平臺:DataFunTalk

導讀:

隨著本地數據遷移上雲、雲上數據交換等多源異構數據源數據同步需求日益增多,傳統通過編寫腳本進行數據同步的方式投入高、效率低、運維管理困難;在公司內部,多款移動雲資料庫和大數據類產品根據客戶需求迫切希望集成數據同步能力,但缺少易用的框架,從0開始研發投入研發成本高。

針對上述問題,基於FlinkX多源異構數據同步框架,實現了用戶自建和移動雲上消息中間件、資料庫、對象存儲等多種異構數據源雙向讀寫,基於社區版實現了On k8s改造,需簡單配置即可滿足用戶數據快速上雲及雲上數據高效交換需求,降低開發運維投入,該成果已在移動雲至少3款產品中應用。

本文的主要內容包括:

FlinkX簡介

功能及原理

雲上入湖改造

展望

一、FlinkX簡介

1. 背景介紹

file

現在市面上有很多種資料庫產品,包括傳統的RDB和大數據相關的NoSQL,一般企業稍微大一點規模都會同時有各種各樣的資料庫。為什麼會有這麼多數據源?是因為不同的數據源適應不同的場景,但這麼多數據源會給開髮帶來困難。

file

傳統的企業業務庫多數還是MySQL,Oracle這種傳統RDB,如果進行簡單的增刪改查是沒有問題的,但遇到大批量的數據計算這些RDB就無法支持了,所以就需要大數據的存儲。但是業務數據又在傳統資料庫中,所以傳統資料庫和大數據之間需要一個同步遷移的工具。

file

FlinkX這個工具相對比較小眾,是袋鼠雲開源的項目。更熟悉的工具可能是Sqoop和阿裡開源的DataX,上圖是一個簡單的對比,我們開始選型的時候也做過調研,包括它的運行模式,插件豐富度,是否支持斷點續傳等功能,特別是我們是做數據湖的,需要對數據湖插件的支持,還有考慮新增插件開發的難度。綜合調研下來,我們最終選擇了FlinkX。多數傳統的企業使用Sqoop比較多,因為他們只會在RDB和大數據之間做遷移,但是Sqoop已經在今年6月份被移除了Apache 頂級項目,上一次更新是在2019年1月份,已經2年多沒有任何的開發更新了,所以這個項目已經沒有新功能開發,這也不滿足我們的需求。之前我們也在移動雲上基於Sqoop做了一個插件,但是發現Sqoop在開發、運行上不太符合我們的場景。最終我們選定了FlinkX這個工具。

2. Flink簡介

file

什麼是FlinkX呢?FlinkX是一個基於Flink流計算框架實現的數據同步插件,它可以實現多種數據源高效的數據同步,基本功能和DataX和Sqoop差不多。

file

批同步方面支持的數據源跟DataX相當,但是在流式同步方面比較有優勢,因為它是基於Flink開發的,所以在流式數據方面支持的數據源比較全,比如Kafka,Pulsar這種消息隊列,還有資料庫的Binlog這種增量更新的數據同步,功能非常強大。基於開源社區1.11版本我們自己又開發了一些插件:對S3的寫入、Hudi數據湖的寫入、對Pulsar的寫入。Pulsar部門已經開源提交到社區了,S3和Hudi暫時還沒有提交。

二、功能及原理

接下來看一下FlinkX的功能和簡單的原理。

1. 斷點續傳

file

首先一個很棒的功能是斷點續傳,當然這個斷點續傳不是針對消息隊列來說的,因為消息隊列天然支持斷點續傳。FlinkX依賴Flink的checkpoint機制,對RDB擴展了斷點續傳的功能。但是它有一個前提,首先是關係型資料庫需要包含升序的欄位,然後是需要支持數據的過濾,最後是需要支持事務。比如使用MySQL時如果需要斷點續傳,配置了這個功能後它會根據欄位做一個取模,然後在數據搬運的過程中當前節點的數據會在checkpoint裡面保存,當需要重新運行任務的時候它會取上一次失敗的節點開始的那個點,因為它還需要根據保存失敗的id的數據做一下過濾,最後再從那個節點重新開始運行任務,這樣數據量比較大的時候會稍微節約一點時間。

2. 指標監控

file

監控方面它會依賴Flink本身的監控功能,Flink內部有一些Accumulators和metric統計指標,如果把它運行在Flink上的話就可以通過Flink的DashBoard來查看Job的狀態。

file

或者是把一些指標數據收集到Prometheus裡面,例如基本的條數,統計的數據量和錯誤的數據量都可以通過Prometheus收集之後再通過Grafana這樣的一些工具做展示。目前線上的這個功能還在開發中。

3. 錯誤統計和數據限制

file

它還有一個比較好的功能是速率限制。當我們讀取數據寫入的時候,很多用戶首先擔心的問題是它會影響到生產庫,因為多數企業的庫可能沒有主從策略,生產庫是單實例運行的。如果這種搬運數據的任務影響到生產庫的話用戶基本上是不能接受的。所以做速率限制的功能對傳統用戶就非常友好。它的速率限制是基於Guava的RateLimit,根據令牌工廠生產令牌的方式做的速率限制,跟另外的漏斗演算法稍微有一點差別。缺點是峰值還是會很高,因為它保證的是平均速率限制在某一個範圍之內。

4. 插件式開發

file

FlinkX的插件式開發模式,與Sqoop和DataX類似,不同的數據源都抽象成一個Reader插件和一個Writer插件,然後整個數據同步的任務和公有的邏輯就被抽象在一個統一的模塊中。一個模塊再根據同步任務的配置載入相應的Reader,Writer最後組裝成Flink任務,並提交到Flink集群去執行。

file

我們可以簡單看一下任務配置,都是基於JSON的方式配置基礎的Reader,Writer,然後是一些綜合的錯誤條數限制和速率限制,這邊的代碼就會根據配置文件通過Reader生成一個Flink Source,再通過Writer生成Sink,熟悉Flink代碼的小伙伴對這塊應該比較熟悉,其實就是Flink從Source端讀數據然後往Sink端寫數據,相對來說比較簡單。

三、雲上入湖改造

雲上入湖這裡我們做了一些改造。

1. K8s

file

首先是K8S的改造,因為社區的1.11版本支持的是Local,Standalone,YARN Session,YARN Perjob的模式,對雲原生方式的開發不是太友好。並且Flink原生的1.12版本已經支持K8S調度運行了,所以我們把基於FlinkX的1.11版本Flink升級到了1.12,讓它原生就可以支持K8S運行,這樣的話對我們任務的彈性擴縮容就更加友好,對入湖的任務資源隔離也比較友好,相互之間沒有影響。這裡也是基於Flink 1.12把裡面的ApplicationClusterDeployer這部分代碼做了一些簡單的改造,來適配我們的一些系統。基本上是把K8S的一些配置組裝一下,然後把FlinkX的一些Jar包的路徑寫進去,最後提交任務到我們的K8S集群。

file

我們的JobManager會通過Quartz來做FlinkX任務的調度,然後通過Flink的客戶端調用K8S的客戶端,最終把任務提交到K8S上去執行。

2. Hudi寫入

file

我們擴展了一個Hudi的插件,因為FlinkX裡面插件非常多,我們這邊會考慮到寫HBase和寫Hive之類的情況,開發過程中遇到了很多Jar包衝突的問題,所以我們給Hudi社區版0.09版本打了非常多的shade Jar,保證我們的線上運行沒有衝突,主要是avro的版本依賴問題。我們這邊HBase和Hive依賴的avro版本跟Hudi的版本會不一致,版本相容性之間有一些問題。

file

這裡看一下Hudi插件預覽的樣子,參考了Hudi源碼裡面加了Client的Example,也就是先載入Hudi配置,初始化表和Hive的配置,最後通過Kafka做實時數據寫入。目前只提供Upsert的支持,後期考慮MySQL Binlog支持的話會增加Delete功能的支持。

3. 日誌

file

還有一個改造可能不屬於FlinkX,就是我們的日誌功能,基於K8S Fluentd的一個小工具,EFK這套系統去收集日誌。整個過程對我們的業務是沒有入侵,沒有感知的,最終我們的日誌解析收集到ES中。Fluentd跟K8S結合的比較好的地方就是它可以採集到NameSpace,PodName, NodeSelector等數據,為後面查詢錯誤日誌提供了方便。

file

上圖就是使用Fluentd收集到的一些Pod的日誌,左側這邊看到有很多K8S的元數據信息,例如ContainerName,鏡像,NodeSelector,PodId等等這些數據。當然這個Kibana是我們留給後端開發用來排錯的,目前給前端用戶展示的還是把原始日誌數據做了彙總之後,通過頁面對應到任務上去查看。

四、展望

最後一部分是我們對於FlinkX的一些展望,先來看一下FlinkX V1.12的一些新特性:

與FlinkStreamSQL融合;

增加了transformer運算元,支持SQL的轉換;

插件向Flink社區看齊,不再區分Reader、Writer,統一命名成Connector,遵循Flink社區的規範,這樣統一以後FlinkX就可以和社區保持相容。理論上在使用FlinkX時可以使用Flink的原生Connector。Flink也可以調用FlinkX的Connector,這樣的話FlinkX就可以做成插件放到Flink的集群裡面,後面對於做湖倉一體或者Server開發就會非常的方便。

數據結構優化

支持二階段提交、數據湖Iceberg和提交kubernetes

對於數據入湖來說,目前的FlinkX有一個缺點,就是只支持結構化數據的傳輸,還不能原生支持二進位文件的同步。如果數據要入湖,會有很多媒體文件,Excel、Word、圖片、視頻等等,這一塊後期可能會自己去開發一些插件支持。

file

升級到1.12後對FlinkSQL的支持會更加友好,這樣傳統的Lambda升級到Kappa架構,對於習慣寫SQL做數據抽取轉換的用戶就非常友好,基本上可以靠一條SQL去實現流批一體化的任務,進一步降低開發維護的難度。我們可以從Kafka讀取一條數據,中間做一些簡單的轉換後寫到MySQL。我們後面資料庫肯定要支持越來越多的實時數據寫入,所以後期用SQL的方式開發這些任務就會更加便捷。

五、問答環節

Q:一般情況下FlinkX作業分配多少CPU和記憶體資源?

A:我們這邊一般定義一個Slot是一核2g,普通的一個MySQL到MySQL這樣的一個任務預設三個併發,用戶更多的是擔心我們的速度太快影響生產庫,目前自定義還沒有開放,後面具體的併發度會開放可以讓用戶自定義,目前Slot是固定的一核2g。

Q:現在流批一體的應用範圍廣嗎?

A:我認為是挺廣的,對於移動集團的一些項目,其實我們在適配他們的一些場景,主要還是基於消息隊列和MySQL的Binlog。我們之前遇到的用戶他在阿裡雲上訂購了結構化數據,現在他想上移動雲,但是他的生產庫又不能斷,他想做這樣的一個遷移,這就是需要流批一體的場景。他需要先做一個批的任務,把他歷史的數據搬運過來,再基於他的Binlog增量訂閱,實時同步更新他的增量數據,這就是一個很典型的傳統用戶的場景。再一個就是有一些大批量數據走Kafka,原始數據還是需要落一份到HDFS,但是需要實時的做一些彙總,這也算是一個比較典型的場景,會做流批一體的任務,我目前主要是針對這兩種場景做一些開發。

Q:FlinkX相較於FlinkCDC優勢在哪裡?

A:單說FlinkCDC他只是支持結構化數據增量更新,FlinkX如果是1.12版本它跟FlinkCDC之間的插件一些是共用的,然後他相較於原生的FlinkCDC做了一些擴展,特別是它會支持很多國產的資料庫,比如達夢,FlinkCDC目前還不支持。任務配置方式的話,FlinkX是基於JSON的,對於寫Flink代碼的的普通用戶更加友好。總結一句話就是擴展了更多插件。

Q:流批一體真的會減少機器的預算嗎?計算資源減少了還是存儲資源減少了?

A:存儲會減少一點,計算可能不會減少,因為流批一體的話,是在用同一套代碼維護批任務和流任務,中間的數據如果沒有必要的話是不用落地的,這塊肯定是節省存儲資源的。計算資源跟原來分開跑的話可能是相當的,不會有明顯的減少,主要是節省了存儲資源。

想瞭解或咨詢更多有關袋鼠雲大數據產品、行業解決方案、客戶案例的朋友,瀏覽袋鼠雲官網:https://www.dtstack.com/?src=szbky

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


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 使用的MCU型號為 AIR32F103CCT6. 通過工作機制和示例代碼, 說明如何使用AIR32自帶的記憶體實現簡單的語音錄製和播放功能, 以及使用 ADPCM 對音頻數據進行壓縮, 提高錄製時長. 通過這些機制, 可以快速擴充為實用的錄製設備, 例如外掛I2C或SPI存儲, 或提升無線傳輸的音質,... ...
  • 1、安裝軟體包時報錯 sudo apt-get install make libc6:i386 lib32z1 lib32stdc++6 zlib1g-dev libncurses5-dev ncurses-term libncursesw5-dev g++ u-boot-tools:i386 te ...
  • powercfg /lastwake 查詢誰喚醒了我的電腦。 禁止設備喚醒電腦:設備管理器——滑鼠和其它指針設備——電源管理——允許此設備喚醒電腦(取消勾選) ...
  • Mac 如何連接遠程伺服器 iTerm2可以以明文保存密碼,實現快速連接遠程伺服器,但是,密碼最好不要暴露在明文里。我們可以藉助更為通用的方法。 在Mac這樣的類UINX系統中,使用iTerm2這樣的終端工具(當然也可以使用Mac自帶的終端)連接遠程伺服器,邏輯是比較通暢的,不需要像Win那樣使用X ...
  • 一、創建表的完整語法 1、創建表的語法 create table 表名( ​ 欄位名1 欄位類型(數字) 約束條件, ​ 欄位名2 欄位類型(數字) 約束條件, ​ 欄位名3 欄位類型(數字) 約束條件,...) 2、創建表的條件 欄位名和欄位類型必須填寫 數字和約束條件是可選的,可以不填 約束條件 ...
  • 我們將 身份驗證 選擇為 SQL Server 身份驗證。輸入剛剛創建的用戶名和密碼: 但是現在還是不能直接使用,因為許可權問題。你會看到下麵的錯誤提示框: 解決問題的方法 所以,我們現在還是需要通過 Windows 身份驗證 連接資料庫。在裡面進行下麵的設置設置。 Step 1 . 對著伺服器,點擊 ...
  • 字元編碼與配置文件 \s查看MySQL相關信息 相關信息有:當前用戶、版本、編碼、埠號 MySQL5.6及之前的版本編碼需要人為統一 之後的版本已經全部預設統一 如果想要永久修改編碼配置 需要操作配置文件my-default.ini 註意事項: mysql預設埠號3306 當重覆起mysql服務 ...
  • 前幾天,Gartner 發佈了企業機構在2023年需要探索的十大戰略技術趨勢。 Gartner 傑出研究副總裁 Frances Karamouzis 表示:“為了在經濟動蕩時期增加企業機構的盈利,首席信息官和IT高管必須在繼續加快數字化轉型的同時,將目光從節約成本轉向新的卓越運營方式。Gartner ...
一周排行
    -Advertisement-
    Play Games
  • 簡介 本文的初衷是希望幫助那些有其它平臺視覺演算法開發經驗的人能快速轉入Halcon平臺下,通過文中的示例開發者能快速瞭解一個Halcon項目開發的基本步驟,讓開發者能把精力完全集中到演算法的開發上面。 首先,你需要安裝Halcon,HALCON 18.11.0.1的安裝包會放在文章末尾。安裝包分開發和 ...
  • 本文是對Datawhale的動手學數據分析課程的學習總結,記錄了整體的學習過程、答案以及個人感想,代碼較為詳細。 ...
  • JZ7重建二叉樹 描述 給定節點數為 n 的二叉樹的前序遍歷和中序遍歷結果,請重建出該二叉樹並返回它的頭結點。 例如輸入前序遍歷序列{1,2,4,7,3,5,6,8}和中序遍歷序列{4,7,2,1,5,3,8,6} 提示: 1.vin.length == pre.length 2.pre 和 vin ...
  • 我們都知道在Java編程中多線程的同步使用synchronized關鍵字來標識,那麼這個關鍵字在JVM底層到底是如何實現的呢。 我們先來思考一下如果我們自己實現的一個鎖該怎麼做呢: 首先肯定要有個標記記錄對象是否已經上鎖,執行同步代碼之前判斷這個標誌,如果對象已經上鎖線程就阻塞等待鎖的釋放。 其次要 ...
  • 目錄 一.OpenGL 色階 1.Windows OpenGL ES 版本 2.Windows OpenGL 版本 二.OpenGL 色階 GLSL Shader 三.猜你喜歡 零基礎 OpenGL ES 學習路線推薦 : OpenGL ES 學習目錄 >> OpenGL ES 基礎 零基礎 Ope ...
  • 1. 查看Linux伺服器版本信息 # cat /etc/redhat-release CentOS Linux release 7.4.1708 (Core) 2. 禪道開源版安裝包下載 wget http://dl.cnezsoft.com/zentao/9.8.2/ZenTaoPMS.9.8. ...
  • Spring 框架可以為 Java 應用程式開發提供全面的基礎設施支持,它是現在非常流行的 Java 開源框架,對於一個 Java 開發人員來說,熟練掌握 Spring 是必不可少的。 ...
  • 前言 本篇是c++總結的第二篇,關於c++的對象模型,在構造、拷貝虛函數上重點分析,也包含了c++11class的新用法和特性,如有不當,還請指教! c++三大特性 訪問許可權 ​ 在c++中通過public、protected、private三個關鍵字來控製成員變數和成員函數的訪問許可權,它們分別表示 ...
  • 一.小結 1.使用二維數組來存儲表格 2.可以使用以下語法來聲明二維數組變數: 元素類型[ ] [ ]數組變數 3.可以使用以下語法來創建二維數組變數: new 元素類型 [行的個數][列的個數] 4.使用下麵的語法表示二維數組中的每個元素: 數組變數[行下標][列的個數] 5.可使用數組初始化語法 ...
  • typimg是一款為typora編輯器提供圖像自定義上傳服務的工具,該工具將在typora中輸入的網路圖片、本地圖片、剪貼板圖片/截圖上傳到博客園,支持在MacOS、Windiws、Linux三個平臺上運行。 ...