ChunJun（原FlinkX）是一個基于 Flink 提供易用、穩定、高效的批流統一的資料集成工具，2018年4月，秉承著開源共享的理念，數堆疊技術團隊在github上開源了FlinkX，承蒙各位開發者的合作共建，FlinkX得到了快速發展，
兩年后的2022年4月，技術團隊決定對FlinkX進行整體升級，并更名為ChunJun，希望繼續和各位優秀開發者合作，進一步推動資料集成/同步的技術發展，
因該文創作于于FlinkX更名為ChunJun之前，因此文中仍用FlinkX來進行分享，重要的事情說三遍：
FlinkX即是ChunJun
FlinkX即是ChunJun
FlinkX即是ChunJun
進入正文分享??????
分享嘉賓：馮江濤 中國移動云能力中心
編輯整理：陳凱翔 亞廈股份
出品平臺：DataFunTalk
導讀：
隨著本地資料遷移上云、云上資料交換等多源異構資料源資料同步需求日益增多，傳統通過撰寫腳本進行資料同步的方式投入高、效率低、運維管理困難；在公司內部，多款移動云資料庫和大資料類產品根據客戶需求迫切希望集成資料同步能力，但缺少易用的框架，從0開始研發投入研發成本高，
針對上述問題，基于FlinkX多源異構資料同步框架，實作了用戶自建和移動云上訊息中間件、資料庫、物件存盤等多種異構資料源雙向讀寫，基于社區版實作了On k8s改造，需簡單配置即可滿足用戶資料快速上云及云上資料高效交換需求，降低開發運維投入，該成果已在移動云至少3款產品中應用，
本文的主要內容包括：
FlinkX簡介
功能及原理
云上入湖改造
展望
一、FlinkX簡介
1. 背景介紹
現在市面上有很多種資料庫產品，包括傳統的RDB和大資料相關的NoSQL，一般企業稍微大一點規模都會同時有各種各樣的資料庫，為什么會有這么多資料源？是因為不同的資料源適應不同的場景，但這么多資料源會給開發帶來困難，
傳統的企業業務庫多數還是MySQL，Oracle這種傳統RDB，如果進行簡單的增刪改查是沒有問題的，但遇到大批量的資料計算這些RDB就無法支持了，所以就需要大資料的存盤，但是業務資料又在傳統資料庫中，所以傳統資料庫和大資料之間需要一個同步遷移的工具，
FlinkX這個工具相對比較小眾，是袋鼠云開源的專案，更熟悉的工具可能是Sqoop和阿里開源的DataX，上圖是一個簡單的對比，我們開始選型的時候也做過調研，包括它的運行模式，插件豐富度，是否支持斷點續傳等功能，特別是我們是做資料湖的，需要對資料湖插件的支持，還有考慮新增插件開發的難度，綜合調研下來，我們最終選擇了FlinkX，多數傳統的企業使用Sqoop比較多，因為他們只會在RDB和大資料之間做遷移，但是Sqoop已經在今年6月份被移除了Apache 頂級專案，上一次更新是在2019年1月份，已經2年多沒有任何的開發更新了，所以這個專案已經沒有新功能開發，這也不滿足我們的需求，之前我們也在移動云上基于Sqoop做了一個插件，但是發現Sqoop在開發、運行上不太符合我們的場景，最終我們選定了FlinkX這個工具，
2. Flink簡介
什么是FlinkX呢？FlinkX是一個基于Flink流計算框架實作的資料同步插件，它可以實作多種資料源高效的資料同步，基本功能和DataX和Sqoop差不多，
批同步方面支持的資料源跟DataX相當，但是在流式同步方面比較有優勢，因為它是基于Flink開發的，所以在流式資料方面支持的資料源比較全，比如Kafka，Pulsar這種訊息佇列，還有資料庫的Binlog這種增量更新的資料同步，功能非常強大，基于開源社區1.11版本我們自己又開發了一些插件：對S3的寫入、Hudi資料湖的寫入、對Pulsar的寫入，Pulsar部門已經開源提交到社區了，S3和Hudi暫時還沒有提交，
二、功能及原理
接下來看一下FlinkX的功能和簡單的原理，
1. 斷點續傳
首先一個很棒的功能是斷點續傳，當然這個斷點續傳不是針對訊息佇列來說的，因為訊息佇列天然支持斷點續傳，FlinkX依賴Flink的checkpoint機制，對RDB擴展了斷點續傳的功能，但是它有一個前提，首先是關系型資料庫需要包含升序的欄位，然后是需要支持資料的過濾，最后是需要支持事務，比如使用MySQL時如果需要斷點續傳，配置了這個功能后它會根據欄位做一個取模，然后在資料搬運的程序中當前節點的資料會在checkpoint里面保存，當需要重新運行任務的時候它會取上一次失敗的節點開始的那個點，因為它還需要根據保存失敗的id的資料做一下過濾，最后再從那個節點重新開始運行任務，這樣資料量比較大的時候會稍微節約一點時間，
2. 指標監控
監控方面它會依賴Flink本身的監控功能，Flink內部有一些Accumulators和metric統計指標，如果把它運行在Flink上的話就可以通過Flink的DashBoard來查看Job的狀態，
或者是把一些指標資料收集到Prometheus里面，例如基本的條數，統計的資料量和錯誤的資料量都可以通過Prometheus收集之后再通過Grafana這樣的一些工具做展示，目前線上的這個功能還在開發中，
3. 錯誤統計和資料限制
它還有一個比較好的功能是速率限制，當我們讀取資料寫入的時候，很多用戶首先擔心的問題是它會影響到生產庫，因為多數企業的庫可能沒有主從策略，生產庫是單實體運行的，如果這種搬運資料的任務影響到生產庫的話用戶基本上是不能接受的，所以做速率限制的功能對傳統用戶就非常友好，它的速率限制是基于Guava的RateLimit，根據令牌工廠生產令牌的方式做的速率限制，跟另外的漏斗演算法稍微有一點差別，缺點是峰值還是會很高，因為它保證的是平均速率限制在某一個范圍之內，
4. 插件式開發
FlinkX的插件式開發模式，與Sqoop和DataX類似，不同的資料源都抽象成一個Reader插件和一個Writer插件，然后整個資料同步的任務和公有的邏輯就被抽象在一個統一的模塊中，一個模塊再根據同步任務的配置加載相應的Reader，Writer最后組裝成Flink任務，并提交到Flink集群去執行，
我們可以簡單看一下任務配置，都是基于JSON的方式配置基礎的Reader，Writer，然后是一些綜合的錯誤條數限制和速率限制，這邊的代碼就會根據組態檔通過Reader生成一個Flink Source，再通過Writer生成Sink，熟悉Flink代碼的小伙伴對這塊應該比較熟悉，其實就是Flink從Source端讀資料然后往Sink端寫資料，相對來說比較簡單，
三、云上入湖改造
云上入湖這里我們做了一些改造，
1. K8s
首先是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集群，
我們的JobManager會通過Quartz來做FlinkX任務的調度，然后通過Flink的客戶端呼叫K8S的客戶端，最終把任務提交到K8S上去執行，
2. Hudi寫入
我們擴展了一個Hudi的插件，因為FlinkX里面插件非常多，我們這邊會考慮到寫HBase和寫Hive之類的情況，開發程序中遇到了很多Jar包沖突的問題，所以我們給Hudi社區版0.09版本打了非常多的shade Jar，保證我們的線上運行沒有沖突，主要是avro的版本依賴問題，我們這邊HBase和Hive依賴的avro版本跟Hudi的版本會不一致，版本兼容性之間有一些問題，
這里看一下Hudi插件預覽的樣子，參考了Hudi原始碼里面加了Client的Example，也就是先加載Hudi配置，初始化表和Hive的配置，最后通過Kafka做實時資料寫入，目前只提供Upsert的支持，后期考慮MySQL Binlog支持的話會增加Delete功能的支持，
3. 日志
還有一個改造可能不屬于FlinkX，就是我們的日志功能，基于K8S Fluentd的一個小工具，EFK這套系統去收集日志，整個程序對我們的業務是沒有入侵，沒有感知的，最終我們的日志決議收集到ES中，Fluentd跟K8S結合的比較好的地方就是它可以采集到NameSpace，PodName, NodeSelector等資料，為后面查詢錯誤日志提供了方便，
上圖就是使用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、圖片、視頻等等，這一塊后期可能會自己去開發一些插件支持，
升級到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=https://www.cnblogs.com/DTinsight/archive/2022/11/23/szbky
同時，歡迎對大資料開源專案有興趣的同學加入「袋鼠云開源框架釘釘技術qun」，交流最新開源技術資訊，qun號碼：30537511，專案地址：https://github.com/DTStack/Taier
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/538225.html
標籤：其他