開源共建 | Dinky 擴展批流統一數據集成框架 ChunJun 的實踐分享

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

一、前言 ChunJun（原FlinkX）是一個基於 Flink 提供易用、穩定、高效的批流統一的數據集成工具，既可以採集靜態的數據，比如 MySQL，HDFS 等，也可以採集實時變化的數據，比如 binlog，Kafka等。同時 ChunJun 也是一個支持原生 FlinkSql所有語法和特性的計 ...


一、前言

ChunJun（原FlinkX）是一個基於 Flink 提供易用、穩定、高效的批流統一的數據集成工具，既可以採集靜態的數據，比如 MySQL，HDFS 等，也可以採集實時變化的數據，比如 binlog，Kafka等。同時 ChunJun 也是一個支持原生 FlinkSql所有語法和特性的計算框架。

ChunJun 具有豐富的插件種類，多達40種，如常見的 mysql、binlog、logminer 等，大部分插件都支持 source/reader、sink/writer 及維表功能。目前很多用戶在思考能否在 Dinky 上使用 ChunJun 的插件以提供更全面的能力。那本文將帶來如何在 Dinky 上集成 ChunJun 豐富的插件，其實簡單，那我們開始吧。

二、部署 Flink+ChunJun

編譯

註意，如果需要集成 Dinky，需要將 ChunJun項目下的 chunjun-core 的pom 文件中的 logback-classic 和 logback-core 註釋掉，否則容易在 Dinky 執行 sql 任務的時候報錯。

file
然後執行：

file

部署

使用 ChunJun 需要先部署 Flink 集群，其部署本文不再做指導。

值得註意的是，如果你需要調用 Flinkx 的 connect jar 的話，則需要將 classloader.resolve-order 改成 parent-first。修改完成配置以後，把 Flinkx 的 jar 包複製過來，主要是 chunjun-clients-master.jar（Flinkx 現在改名 ChunJun ）以及 chunjun 的其它 connector 放到 flink/lib 目錄下，如圖所示。
file

異常處理

如果啟動集群時出現異常，即 Flink standalone 集群載入 flinkx-dist 里 jar 包之後，集群無法啟動，日誌報錯：Exception in thread "main" java.lang.NoSuchFieldError: EMPTY_BYTE_ARRAY.

Exception in thread"main"java.lang.NoSuchFieldError:EMPTY_BYTE_ARRAY
at org.apache.logging.log4j.core.config.ConfigurationSource.(ConfigurationSource.java:56)
at org.apache.logging.log4j.core.config.NullConfiguration.< init>(NullConfiguration.java:32)
at org.apache.logging.log4j.core.LoggerContext.< clinit>(LoggerContext.java:85)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:264)
at org.apache.log4j.LogManager.< clinit>(LogManager.java:72)
at org.slf4j.impl.Log4jLoggerFactory.getLogger(Log4jLoggerFactory.java:73)
at org.slf4j.LoggerFactory.getLogger(LoggerFactory.java:285)
at org.slf4j.LoggerFactory.getLogger(LoggerFactory.java:305)
at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.< clinit>(ClusterEntrypoint.java:107)

原因：這個報錯是因為 log4j 版本不統一導致的，因為 flinkx-dist 中部分插件引用的還是舊版本的 log4j 依賴，導致集群啟動過程中，出現了類衝突問題；

方案：臨時方案是將 flink lib 中 log4j 相關的jar包名字前加上字元 ‘a‘，使得flink standalone jvm 優先載入。
file

三、部署 Dinky

編譯

file
編譯完成後的壓縮包在 Dinky 根目錄下的 build 文件夾下。

部署

1、上傳dlink壓縮包到部署伺服器

2、解壓

file
3、資料庫初始化

4、把 flink 的 jar 放到 dlink 目錄下

file

因為目前 flinkx 的穩定版本是 1.12.7，所以我們把 dlink 預設的 client 版本修改為 1.12

file
lib下的目錄如圖：

file
註意：因為我沒有用上 dlink-connector-jdbc 的 jar 包，所以圖中的 dlink-connector-jdbc-1.13-0.6.4-SNAPSHOT.jar 沒有換成1.12版本的，可以去掉。

啟動

啟動命令
file

註冊集群實例

在集群實例中註冊已經啟動的 Flink 集群。
file

四、示例分享

添加依賴

這裡演示 mysql->mysql 的同步作業，所以需要 Flinkx 的 mysql-connector.jar 以及核心 jar。
file

編寫作業

Mysql DDL：

CREATE TABLE datasource_classify (
id int unsigned NOT NULL AUTO_INCREMENT COMMENT '自增id',
classify_code varchar(64) NOT NULL COMMENT '類型欄唯一編碼',
sorted int NOT NULL DEFAULT '0' COMMENT '類型欄排序欄位 預設從0開始',
classify_name varchar(64) NOT NULL COMMENT '類型名稱 包含全部和常用欄',
is_deleted tinyint NOT NULL DEFAULT '0' COMMENT '是否刪除,1刪除，0未刪除',
gmt_create datetime DEFAULT CURRENT_TIMESTAMP,
gmt_modified datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY classify_code (classify_code)
) ENGINE=InnoDB AUTO_INCREMENT=12 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='數據源分類表';

Flink Sql:

CREATE TABLE source
(
id bigint,
classify_code STRING,
sorted int,
classify_name STRING,
is_deleted int,
gmt_create timestamp(9),
gmt_modified timestamp(9),
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'mysql-x',
'url' = 'jdbc:mysql://192.168.31.101:3306/datasource?useSSL=false',
'table-name' = 'datasource_classify',
'username' = 'root',
'password' = 'root'
,'scan.fetch-size' = '2'
,'scan.query-timeout' = '10'
);

CREATE TABLE sink
(
id bigint,
classify_code STRING,
sorted int,
classify_name STRING,
is_deleted int,
gmt_create timestamp(9),
gmt_modified timestamp(9),
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'mysql-x',
'url' = 'jdbc:mysql://192.168.31.106:3306/test?useSSL=false',
'table-name' = 'datasource_classify',
'username' = 'root',
'password' = 'root'
,'scan.fetch-size' = '2'
,'scan.query-timeout' = '10'
);

insert into sink
select *
from source u;

執行任務

file
選中 Yarn Session 模式提交作業。
file
提交後可從執行歷史查看作業提交狀況。
file
進程中可以看的 Flink 集群上批作業執行完成。

對比數據

源庫：
file
目標庫：
file
同步成功，很絲滑。

五、總結

在集成 ChunJun 的時候遇到的問題大部分都是缺包以及包衝突，所以只需要註意一下這個問題就能比較好的進行集成。

在集成服務的時候建議是，先把 Flink 和 ChunJun 進行集成，確保服務能夠正常啟用以後再進行 Dinky 的集成，這樣有利於快速定位查找問題，如果遇到文章之外的問題，也可以查看 Dinky 官網FAQ | Dinky (dlink.top) chunjun的官網QuickStart | ChunJun 純鈞 (dtstack.github.io/chunjun/)，看看是否有類似問題的解決辦法作為參考。

六、用戶體驗

因為本人目前還是處於學習使用的過程中，所以很多功能沒有好好使用，待自己研究更加透徹後希望寫一篇文章，優化官網的用戶手冊。以下的優缺點以及建議都是目前我在使用學習的過程中遇到的問題。

優點

Dinky 最吸引我的地方應該就是 sql 編輯模版了，直接快捷鍵生成 sql 模版，在開發測試中屢試不爽。在集成了 ChunJun(Flinkx) 以後，能夠做到多源數據的離線跑批任務及日常小批量實時任務的同步。支持各種類型的任務執行方式。

缺點

ui 上適配還有點小問題，例如：打開 F12 調整寬度後，再關閉，頁面 ui 不會自適應，需要刷新。

期待改進點

1、更多的自定義異常、業務異常

2、增加新的嚮導模式，結合數據源，通過 webUI 可以一鍵引入欄位或者勾選需要的欄位，生成 Flink Sql 的一大部分配置

CREATE TABLE 表名
(
-- 頁面勾選欄位，欄位從元數據直接拉取
id bigint,
classify_code STRING,
sorted int,
classify_name STRING,
is_deleted int,
gmt_create timestamp(9),
gmt_modified timestamp(9),
PRIMARY KEY (id) NOT ENFORCED
) WITH (
-- 從選擇的數據中獲取
'connector' = 'mysql-x',
'url' = 'jdbc:mysql://192.168.31.106:3306/test?useSSL=false',
'table-name' = 'datasource_classify',
'username' = 'root',
'password' = 'root'
,
-- 其它非主要配置有用戶自己填寫
);
3、sql 歷史版本管理，目前我已經提交 Feature 並被合併到 0.6.5 版本中。

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

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


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

-Advertisement-
Play Games
更多相關文章
  • java基礎篇—基礎語法
    一、關鍵字和保留字 1.什麼是關鍵字？ 通俗來說就是帶有特殊含義的字元，有自己專門用途的單詞 2.特點? 關鍵字全部由小寫構成，以下是java官方列舉出的關鍵字 註意： 保留關鍵字：指的是現有版本中尚未使用，後續可能會作為關鍵字使用，自己命名標識符要避免使用這些關鍵字，並且Java 的 null 不 ...
  • 資料庫可視化工具分享 （DBeaver）
    前提：最近公司下發通知，所有開發人員 必須 卸載 Navicat 資料庫可視化工具，不知道兄弟們有沒有在使用的，可能現在的反應跟我一樣，一臉懵逼，Navicat為什麼不能使用呢？ 有事沒事找度娘，於是我。。 （簡單粗暴） 搜到了答案。。 原來如此，那就廢話不多說，這邊介紹 使用 DBeaver 來替 ...
  • 猴子補丁
    作用：隨時修改代碼 (在函數或類定義完成之後，再去修改函數的實現過程) """類似猴子補丁在函數定義好之後，再去更改他的行為"""import typesclass Valley: def func(self): return "等待宣告"def common(self): return "只有永不 ...
  • 《Go 語言併發之道》讀書筆記（三）
    1. 擴容方案剖析 1.1 擴容問題 在項目初期，我們部署了三個資料庫A、B、C，此時資料庫的規模可以滿足我們的業務需求。為了將數據做到平均分配，我們在Service服務層使用uid%3進行取模分片，從而將數據平均分配到三個資料庫中。 如圖所示： 後期隨著用戶量的增加，用戶產生的數據信息被源源不斷的 ...
  • Python工具箱系列(十四)
    上文介紹了命令行方式來對文件進行加解密操作。本文將繼續在此基礎上，實現一個快速簡易的GUI界面方便操作，先上代碼看效果。 ...
  • 使用Fody時,CS-SCRIPT動態代碼無法找到程式集
    aspnetcore上傳圖片也就是上傳文件有兩種方式，一種是通過form-data，一種是binary。 先介紹第一種form-data: 該方式需要顯示指定一個IFormFile類型，該組件會動態通過打開一個windows視窗選擇文件 及圖片。 postman演示如上，代碼如下： [HttpPos ...
  • FreeRtos於嵌入式環境的應用
    FreeRtos操作系統 首先，應該介紹什麼是FreeRtos，他於單片機而言就是一個管理器，作為管理者管理嵌入式晶元中的任務，堆棧，中斷，隊列等等資源，對於操作系統而言，又分為實時操作系統和非實時操作系統，實時操作系統代表任務或者某個功能必須在指定的運行時間內完成，保證設備想要執行的功能能立即得到 ...
  • Centos 7 部署 openGauss 3.1.0 級聯備庫集群及集群縮容刪除節點-運維指南
    摘要：先通過OPS確認節點狀態是否已經恢復，或登錄後臺執行cm_ctl query -Cv確認集群是否已經Normal。 本文分享自華為雲社區《【實例狀態】GaussDB CN服務異常》，作者：酷哥。 確認節點狀態 先通過OPS確認節點狀態是否已經恢復，或登錄後臺執行cm_ctl query -Cv ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類