FFA2024分論壇-數據集成.pdf

1、Flink CDC YAML:面向數據集成的 API 設計Flink CDC YAML:The API designed for Data Integration徐榜江（雪盡)Flink PMC Member&Committer，Flink CDC 項目負責人Flink CDCYAML APITransform+AICommunityFlink CDCFlink CDC 項目介紹，定位，優勢Flink CDCFlink CDC 是基于數據庫的日志 CDC（ChangeChange DataData CaptureCapture）技術，實現了全增量一體化讀取的端到端流式數據集成框架，配合 Fli

2、nk 優秀的管道能力和豐富的上下游生態，Flink CDC可以高效實現海量數據的實時集成。PaimonFlink CDC實時實時,一致性一致性Flink CDC全量數據全量數據增量數據增量數據TiDBHologresClickHouseIcebergPaimonSELECTWHEREJOINTop-NFlink SQL APIGROUP BYINSERTmapfilterjoinFlink DataStream APIkeyByflatMapSchema EvolutionSchema SyncSELECTFilterCDC YAML APIFull DB SyncUDFaggregateFl

3、ink CDC 用戶APITiDBRDS MySQLHudi傳統 CDC 數據集成作業DataX/Sqoop全量同步Debezium/Canal增量同步定時合并結果表增量表全量表鏈路組件多鏈路組件多數據一致性數據一致性數據新鮮度數據新鮮度研發技術棧研發技術棧!基于 Flink CDC 的數據集成作業Canal/Debezium增量同步DataX/Sqoop全量同步CDC SourceSinkCustom Logics定時合并結果表增量表全量表全增量一體化不丟不重亞秒級延遲端到端作業!結果表TiDBHologresIcebergPaimonSELECTWHEREJOINTop-NFlink SQ

4、L APIGROUP BYINSERTmapfilterjoinFlink DataStream APIkeyByflatMapSchema EvolutionSchema SyncSELECTFilterCDC YAML APIFull DB SyncUDFaggregateFlink CDC 用戶APITiDBRDS MySQLHudiStarRocks4545端到端 Data Pipeline一個 YAML 文本，一個 Data Pipeline一行 Shell 命令,一個 Flink 作業細粒度 Schema Evolution表結構變更自動同步，無需作業啟停根據容錯和演進策略，支持細

5、粒度控制強大的 Transform 支持支持 SELECT、WHERE、計算列、表達式提供豐富的內置函數，同時支持 UDF全增量一體化全量和增量自動銜接，無鎖算法保證一致性無需調度系統和手工介入Flink CDC 的優勢YAML APIYAML API 設計背景，設計考量，核心特性Flink CDC 發展歷史2020/07 Kick Off2021/082.0 版本 MySQL CDC，Postgres CDC 連接器2022/112.3 版本MySQL CDC 實現增量快照算法2023/103.0 版本 YAML API,端到端流式數據集成框架2024/01 捐贈 ASFYAML 支持 Tr

6、ansform(Projection,Filter,UDF)2024/093.2 版本增量快照框架,覆蓋重點連接器作為Apache Flink 子項目捐贈給 ASFFlinkFlink CDCCDC 1.x1.x：FlinkFlink 的的 CDCCDC 連接器連接器(2)Scan Snapshot Data of table(4)Append Changelog of tableJDBC connectionBinlog connection(1)Lock table for data consistency(3)Release table lock after scanFlinkFlink

7、 CDCCDC 2.x2.x：生產大規?？捎玫模荷a大規?？捎玫?CDCCDC 連接器連接器DBJDBC conJDBC conJDBC conbinlog con原創增量快照算法，解決生產可用問題！無鎖一致性切換Task1Task2Task3分布式快照讀取增量數據讀取自動切換chunk2chunk3chunk1Task1FlinkFlink CDCCDC 1.x1.x&2.x2.x：簡化：簡化 FlinkFlink ETLETL 作業作業Transform(T)Load(L)Extract(E)TiDBClickHouseIcebergHudiHologresFlinkFlinkDebezi

8、umTiDBClickHouseIcebergHudiHologresMySQL CDC SourceFlinkFlink CDCCDC 3.x3.x ：端到端實時數據集成框架：端到端實時數據集成框架自動處理極致擴展版權中立端到端體驗AI/MLAnalytics/BIDatabaseData LakeData WarehouseFlink CDCYAML API 設計PaimonStarRocks 目標用戶：初級用戶，無需專家 設計原則：足夠簡單，足夠強大YAML API 設計mmysqlysql-toto-starrocks.starrocks.yamlyaml$flink-cdc.sh m

9、ysql-to-starrocks.yaml一條命令，一個任務一個文本，一個作業Flink CDC 架構Flink RuntimeFlink CDC RuntimeCDC SourcesSQL APIYAML APIYAML SinksCDC client(TiDB,MongoDB)Debezium(MySQL,PG,Oracle.)DataStream API RowData Delete Insert Update Before Update AfterFlink SQL 作業 DataChangeEvent Delete Insert Update BeforeAfter CreateT

10、ableEventSchemaChangeEvent AddColumnEvent TruncateTableEvent CDC YAML 作業SQL 作業 vs YAML 作業SQL 作業 vs YAML 作業Flink SQL 作業CDC YAML 作業手寫冗雜 DDL create table!無法處理 上游表結構變更!破壞原始 Update Changelog 語義!每張表需要一個算子 TableSource/TableSink!表結構自動發現，整庫同步細粒度 Schema Evolution 支持 原始 changelog 同步單算子讀寫多表DataStream 作業 vs YAML

11、 作業 StreamRecordFlink DataStream 作業 DataChangeEvent CreateTableEventSchemaChangeEventCDC YAML 作業op(user,STRING)(id,INT)(user,STRING)(id,INT)+ILeonard1 op(user,STRING)(id,INT)ULeonard13 op(user,STRING)(id,INT)-DLeonard3 +ILeonard1 -DLeonard3BinaryData+SchemalessULeonard13DataStream 作業 vs YAML 作業Flink

12、 DataStream 作業CDC YAML 作業Java/Scala 專家，分布式編程!DataStream API，State,Checkpoint!Maven 基礎、依賴管理!作業難以復用!YAML 能構建強大的作業，同時屏蔽底層細節 YAML 語法非常容易理解和掌握復制已有作業，簡單修改就能復用面向普通用戶，無需豐富的開發經驗SchemaChangeEventDataChangeEventFlushEvent!Sink notifies flush complete Schema registry applies schema change#Schema operator receiv

13、esSchemaChangeEvent$Schema operator registers schema changethen wait for response(hold upstream).Blocks if SchemaRegistry is busy%Schema registry accepts schema change,and rejects following requests&Schema operator broadcasts FlushEvent,requests registry again to wait for flush complete Schema regis

14、tryconfirms schema evolution completes,ready for next request(Schema operatorreleases upstream$ther schema operator must wait until other schema changes are completesSchema OperatorDataSinkPost PartitionerDataSourceSchemaRegistryMetadataApplierAPPLYINGIDLEFINISHEDWAITINGYAML 核心特性：Schema Evolution YA

15、ML 核心特性：細粒度 Schema Evolution 修改操作修改列名，修改列定義等新增操作新增表，新增列等刪除操作刪除列，刪除表等清空操作清空表數據EXCEPTIONIGNORELENIENTEVOLVETRY_EVOLVE演進容錯Transform+AICDC+RAG 場景，AI Model 設計數據加工(T)數據裝載(L)數據攝取(E)YAML 核心特性：Transform 設計PaimonStarRocksFlink CDC選擇列:SELECT自定義函數：MyUDF計算列:val AS age-18條件過濾:WHERE val 1聲明主鍵：primary key聲明 model:E

16、mbeddingYAML 核心特性：Transform 實現SQL 解析(Calcite)動態編譯(Janino)邏輯表達(YAML)l 列選擇l 條件過濾l 自定義函數l l 注冊表結構l 注冊內置函數l SQL 語法解析l l 表達式生成l 表達式復用l 運行時調用l YAML 核心特性：Transform 實現Data SourceSchema OperatorData SinkPreTransformPostTransform刪除投影無關列生成計算列條件過濾數據行原始 Schema削除 Schema完整 SchemaPaimon完整 SchemaYAML 核心特性：AI model 集

17、成業務場景：知識庫廣告營銷 金融 技術應用：RAG(檢索增強生成)實時投放實時風控StarRocksElasticseachYAML 核心特性：AI model 設計兼容閉源開源&閉源 多種模型Chat&Embedding靈活配置 自定義模型參數代碼復用底層實現復用UDFYAML 核心特性：AI model 實現FLINK-36525 Support for AI Model Integra=on for Data ProcessingOne More ThingCommunity開源社區動態，未來規劃Flink CDC 社區動態Contributors100+Commits1000+Star

18、s5000+02004006008001000120020232024109971454%0306090120150202320241421010100020003000400050006000202320245504429228%41%Flink CDC 社區規劃支持 AI Model（已有PR）支持 Batch pipeline對接更多上下游，如：Iceberg，ClickHouse支持更多類型，包括 schema change 類型 和 event change 類型 解鎖更多新場景 穩定性持續打磨數據限流，細粒度的異?？刂萍嫒莞嗟?Flink 版本，JDK 版本升級至11+更高的三方

19、庫版本，如：Debezium,OBLogproxy,TiCDC遵循 Apache Flink 的貢獻步驟擁有獨立的文檔網站:https:/nightlies.apache.org/flink/flink-cdc-docs-stable 在 Flink 社區郵件列表討論:devflink.apache.org/userflink.apache.org 在 Apache JIRA 管理需求和缺陷:https:/issues.apache.org/jira在 GitHub 上提交代碼PR:https:/ Apache Flink 最活躍 的子項目，Flink CDC：歡迎加入開源社區Flink CD

20、CTHANK YOU實時數據采集在杭州銀行的應用實踐歐陽武林 杭州銀行實時數據負責人背景解決方案展望背景杭州銀行概況40杭州銀行成立于1996年9月，是一家典型的城市商業銀行，堅持服務區域經濟、中小企業和城鄉居民的市場定位，致力于為客戶提供專業、便捷、親和、全面的金融服務2016年10月27日，在上海證券交易所成功掛牌上市，截至2023年底，總資產18413.42億元，2023年營業收入350.16億元，利潤收入162.82億元。環渤海灣長三角珠三角合肥北京浙江省內深圳南京寧波上海全行擁有273家分支機構網點覆蓋長三角、珠三角環渤海灣等發達經濟圈省內地市實現網點全覆蓋設立了15家異地分行l 在

21、英國銀行家雜志 2023年公布的全球銀行1000強排名中，按一級資本排名位列第132位l 2023年財富中國500強企業位列第380位l 連續蟬聯“十佳城商行”l 最具競爭力中小銀行l 最佳普惠金融成效獎l 浙江省融資暢通工程“突出貢獻獎”l 最佳科技金融服務城商行l 支持浙江省經濟社會發展優秀單位一等獎l 獲評“杭州市模范集體”2017202320242007數倉建設在Oracle內部完成，大多數數據在Oracle到Oracle之間同步和計算。DataX隨著數據應用場景變得復雜，數據倉庫上下游需要支持多種數據源，我們開發了第二代數據采集平臺，引擎引入阿里開源的DataX初探Flink CDC

22、數據時效性的要求增加，用戶不斷的提出需要數據實時采集，我們嘗試引入Flink CDC增強Flink CDC構建流批一體的數據采集平臺，并推廣大規模的使用。數據采集的發展歷程DBlink痛點數據時效性差離線/實時割裂我們離線采用DataX實現，實時采用的是Flink CDC，需要維護兩套引擎，兩套采集平臺，實時只支持mysql cdc的采集，缺乏更多的采集源支持。缺乏租戶隔離目前平臺是單租戶模式，需要專門的人來做數據采集的工作，隨著金融行業數字化改革的進展，導致我們數據采集配置的工作負擔特別重，效率不高。目前離線采集作業1w+，部分表的采集時間需要1小時以上，需要更快的同步數據，提高數據產生的時

23、效性。在以DataX為引擎的時候，我們數據都會緩存到Oracle，后續所有的數據直接CDC入湖，縮短數據鏈路數據鏈路冗余解決方案流批一體數據采集平臺架構作業總覽Dataharbor平臺實時分發元數據管理數據對比離線同步實時同步標志位管理離線調度引擎DAG管理定時調度失敗策略補數實時引擎Streampark（Flink管控）作業運維作業開發監控管理告警管理元管理技術底座離線分發核心功能可用性保障作業監控報警管理系統管理作業監控租戶管理FlinkFlink CDCPaimonyarnhdfsCDC支持高可用源數據MYSQLPostgresTiDBOracle目標系統PaimonGuassDBDor

24、isAnalyticDB.TiDBPGSQLMYSQLORACLEPaimonFlink CDCCase1:基于watermark自動生成TagCase2:日結標志位出發生成TagDS(tag產生)HDFSHIVEDorisSpark標簽中心特征中心事件中心報表中心ADB/Doris批流DS(tag消費).流讀批+流分發：數據入湖Flink CDC使用DS調度基于金融日結標志位產生增量快照或者基于上游的eventTime自動生成增量快照；數據出湖，批場景使用DS掃描上游表的tag進行調度，流場景直接flink消費paimon的changelog；流批一體數據湖架構tag1tag2tag3支持F

25、link CDC Yaml APIFlink CDC Yaml API 支持表結構自動同步、數據轉換、整庫同步等高級特性；SQL和Yaml語法都比較簡單，我們改造Streampark SQL作業接口，后端開發cdc client jar來對接flink cdc cli，實現支持Yaml語法。CDC YamlStreamparkFlink SQLShip Client JarCDC Client JarFlink SQL Client JarDeployCluster解析啟動參數Flink解析Yaml啟動TM解析啟動參數解析SQL啟動TMCDCSQLFlink CDC cliFlink CDC

26、commonFlink CDC composeDataHarbor支持Flink Catalog這功能已經貢獻到StreamPark社區，我們實現了一個JDBC Catalog Store Plugin，Streampark通過Flink FactoryUtil進行SPI加載StorePlugin進行Catalog/Database/Table的CRUD，在啟動Flink作業的過程Ship StorePlugin到Flink作業中，Flink通過Catalog的懶加載機制從Catalog Store中獲取Catalog信息。MYSQLDataharbor平臺PGSQLFlink lazy in

27、itialization of catalogsPaimonAPIStreamparkCatalogServiceCatalogBaseLoad StorePluginFactoryUtilCatalog/Database/TableStart JMParser SQLCatalogManagerLoad StorePluginRead CatalogJDBC Catalog StoreShip Catalog Store PluginMYSQL支持Flink Catalog多租戶體系建設業務方數據開發方源系統方采集平臺方開始提出業務需求分析需求，庫表字段需求在catalog list搜索到所

28、需源庫表（部署初始化匯聚）否根據相關信息，在DH注冊元數據、指定庫技術負責人DB技術負責人為table指定技術負責人table技術負責人元數據信息table技術負責人創建入湖數據作業運維:數據對比/日志/重啟已有catalog、申請入戶,提供時效、tag需求擁有湖內表分發權限已入湖表分級等級是否較高提起審批是table技術負責人審批數據架構部經理審批創建分發作業否否是已入湖業務驗收結束作業運維:數據對比/日志/重啟Flink TiDB CDC Connector現狀TiDBTiDBTiKVTiKVTiKVmysqlpaimonAD BPDGrpcTikv scan全量問題1:TiDB協議目前

29、tikv java-client只支持tidb 5.x，高版本Tidb采集的時候報不兼容錯誤。TiDB Java ClientTiDB CDC ClientCDC訂閱增量Flink CDC問題2:CDC Clientjava-client 3.2版本之后就不再維護 cdc部分代碼，代碼穩定性差。問題3:Java Client增量數據采集使用java client 掃描tikv表數據，沒限流操作，導致Tikv節點異常。問題4:Flink框架Flink CDC 1.0框架全量只支持單并發采集，性能差，全量數據全緩存在內存，大表采集非常消耗資源Flink TiDB CDC Connector重構Ti

30、DBTiDBTiKVTiKVTiKVmysqlpaimonAD BPDScanGrpcGrpcGrpc增量流讀MYSQL JDBC（全程無鎖、水平擴展、checkpoint）+基于Flip-27和Dblog paper重構TiDB CDC Connector，實現全程無鎖、水平擴展、checkpoint；全量部分數據通過MYSQL JDBC多并發獲取數據，增量部分通過Grpc訂閱TiKV的binlog日志。增量部分數據采集主要是參照TiFlow中的TiCDC的源碼實現。StreamTiDB CDC ClientTiDB Java ClientTiDB CDC Client 重構（增量數據）Ti

31、DBTiDBTiKVTiKVTiKVinputChinputChinputChRegion workerRegion workerRegion workerMatcherCDC Client APIPDGrpcGrpcGrpcTiDB java-client 3.2 版本之后cdc部分沒有維護了，協議只支持TiDB 5.x版本，沒有異常處理。我們升級java-client的版本，并重構CDC部分的代碼，實現并發采集、事務處理和異常處理。RegionWorker對應TiKV store，負責數據讀取并推送到Matcher中，Matcher進行事物組裝，Sorter對數據進行緩存和排序，CDC C

32、lient對上層暴露接口。Java ClientSorterTransaction CacheRocksDB依賴MatcherMatcher數據入湖/分發管理入湖分發監控與報警展望鏈路穩定性增加全鏈路監控/報警能力，保障數據鏈路的穩定性數據校驗增加字段級的數據校驗能力，校驗入湖數據和源數據的一致性。元數據管理建統一元數據管理層，支持多引擎、權限管控、數據血緣等資源優化減少資源的消耗，使用Flink CDC Yaml的多表采集，AI進行Flink作業資源優化數據探查構建數據探查能力，快速探查入湖數據。數據糾正提供一鍵數據修復，數據異?？焖倩謴驼雇脚_數據逐步演進到湖倉體架構THANK YOU謝

33、謝 觀 看Fink X OceanBase 實現海量數據的快速遷移和實時分析王赫OceanBase 研發工程師01OceanBase 特性介紹02Flink X OceanBase 數據遷移方案介紹0304未來展望Flink X OceanBase 實現實時數據分析OceanBase 特性介紹完全自主研發的數據庫，才能做到核心替代的真正落地自研一體化架構兼容經典模式，實現單機和分布式、TP和AP的融合 原生分布式全量數據校驗真正實現數據強一致，數據不丟失，高可用，平滑擴展單機分布式一體化自研一體化架構突破高性能和高可用，實現應用無限擴展和服務永遠在線 HTAP一份數據既能做事務處理又能實時分析

34、，通過 HTAP 助力拓展更多可能 MySQL平滑遷移業務少量修改甚至不改即可遷移到OB，自動評估和遷移工具 低成本基于 LSM-Tree 的高壓縮引擎平衡了“性能”和“壓縮”的瓶頸，有效降低存儲成本 70%-90%多租戶資源隔離按需使用，靈活管理，適合微服務架構和 SaaS 行業應用原生多租戶架構Oracle 兼容MySQL 兼容 HTAP 引擎（TP+AP)SQL 優化引擎集中式/分布式一體化架構并行執行引擎存儲過程基于 Paxos+數據同步的靈活的容災架構分布式存儲分布式事務分布式調度單機房三副本同城三機房雙機房主備兩地三中心三地五中心靈活的部署模式物理部署專有云/混合云公有云/多云Oc

35、eanBase 4.0 特性總覽4.x 新特性：一體化存儲引擎（行存+列存）Multiple storage versionsDump SSTableMemoryDiskIn-Memory HashIn-Memory B!-TreeMemTableRow-levelIn-MemoryRedo/MVCCMajor SSTable(列存+行存)Block CacheMutationLogsScanQuery基線行存/列存/行列+列存增量行存:MemTable/轉儲合并純列存表/行列冗余表語法說明each column：表中的每一列分別使用列格式存儲純列存表create table t2(pk in

36、t,c1 varchar(1),c2 varchar(1)with column group(each column);語法說明all columns：一個 column group 里包含表里所有的列，即原有的行格式存儲（AP 場景常用于點查、維表關聯、實時更新、主鍵掃描等）each column：表中的每一列分別使用列格式存儲行列冗余表create table t1(pk int,c1 varchar(1),c2 varchar(1)with column group(all columns,each column);行存/列存索引-索引表也可以指定行存、列存說明列存索引應用場景：將行存表

37、中的若干列轉化成列存，用于分析。create table t2(pk int primary key,c1 varchar(1),c2 varchar(1);create index idx1 on t2(c2)with column group(each column);對普通表創建列存索引邏輯結構主表（行存）索引表（列存）說明行存索引應用場景：將列存表中的若干列轉化成行存，用于高性能點查。create table t2(pk int primary key,c1 varchar(1),c2 varchar(1)with column group(each column);create in

38、dex idx1 on t2(c1)with column group(all column);對普通表創建行存索引邏輯結構主表（列存）索引表（行存）4.x 新特性：物化視圖定義：一種特殊視圖，存儲預先計算的結果集。目的：加速查詢，減少系統資源消耗（CPU、網絡等），利用空間換時間。常應用于：數據匯總：匯總每天、每周或每月的銷售數據、統計用戶行為數據等；統計信息報表數據生成：報表系統需要定期生成固定格式的數據報告；復雜查詢加速：大數據量的復雜JOIN、分組和聚合操作比較耗時耗資源，可以將結果物化避免查詢重復計算；監控數據的預聚合：監控數據會按照不同的時間間隔進行展示。lHTAP:一個租戶內同時

39、服務 TP 和 AP 請求 TP 和 AP 共享一份數據 保證復雜的 AP 請求不影響 TP 的業務lAP業務:多個AP業務的處理優先級不同l基于資源組的資源隔離方案 User 級/后臺任務級/SQL級資源隔離，權重設置 CPUs 通過 Cgroup 隔離 Iops 通過 IO 調度算法進行隔離基于資源組的資源隔離方案業務系統業務系統業務系統.資源組資源組資源組.4.x 新特性：大幅增強的資源隔離4.x 新特性：旁路導入MEMTSSTTablet導入數據SSTTabletlockedlockedOSS本地文件路由和控制節點數據節點數據節點數據節點備節點clogobloadertableAPIi

40、nsert into selectload data infile快速執行路徑優化寫入放大問題表鎖阻塞寫入備節點cloginsert/*+append enable_parallel_dml parallel(3)*/into t1 select;load data/*+direct(true,1024)parallel(16)*/infile;旁路導入大幅提升寫入性能5160196773725149201952140245901000200030004000500060004xlarge(16c32G)12xlarge(48c96G)堆表 普通插入堆表 旁路導入索引組織表 普通導入索引組織表

41、 旁路導入導入性能提升 x74.x 新特性：OBKV 多模 APIOceanBase 集群事務層存儲層事務層存儲層事務層存儲層業務節點1.AppOBKV APIHBase 模型Redis 模型時序（CeresDB）圖（TuGraph）通用的使用場景通用的使用場景 通用性：其它模型易于向KV模型轉換 典型使用場景：多模數據服務的高可靠存儲底座 可擴展的元數據存取服務 極致低時延的數據訪問服務OBKV 數據庫集群OBKV Client 多模APIDocument增量操作GIS計算對象存儲接口Json計算GIS計算Lob存儲 不僅僅是不僅僅是KVKV“Value”更豐富：提供Json等標準類型的原生

42、操作接口?！癡alue”操作更高效：提供Json/對象存儲的增量操作等接口。業務節點2AppOBKV APIOBKV Client 多模APIOBKV 數據庫 易用的分布式存儲易用的分布式存儲 簡單易用：基于簡單API操作數據庫 低時延：ByPass SQL&客戶端路由 高可靠：RPO=0，RPO8 可擴展：橫向擴展，自動均衡 低成本：存儲極致壓縮比4.x 新特性：向量檢索能力用戶訪問層多模數據層檢索層索引層內核SQL 多語言客戶端Python SDK基礎類型JSONGISVectorHBaseRedis融合查詢向量索引全文索引GIS空間索引多值索引二級索引事務高可用多分區水平擴展高效存儲彈性

43、擴縮容讀寫分離并行執行兼容 MySQL/Oracle 協議的向量數據庫建表語句，向量作為一個單獨的列create table t1(c1 vector(3),c2 int,c3 float,primary key(c2);數據導入，使用標準SQL語法導入insert into t1 values(1.1,2.2,3.3,1,1.1),(9.1,3.14,2.14,2,2.43);insert into t1 values(7576.42,467.23,2913.762,3,54.6);DML 操作和普通表一致select*from t1;delete from t1 where c2=1;使用

44、DDL進行SQL語法，可以指定索引類型和距離算法CREATE INDEX vidx_c1_t1 on t1(c1 l2)using hnsw;帶有向量計算的SQL，優化器會自動使用向量索引select*,c1 3,1,2 as dis from t1 order by c1 3,1,2;select c3 from t1 order by c1 3,1,2 limit 2;提供 SQL 客戶端和 Python SDK 兩種使式性能強勁的向量數據庫ANN-Benchmarks GIST-960 排名第一 在 90%以上的召回率區間，QPS 超此前最優算法 glass 100%，超基線算法 hns

45、wlib 300%螞蟻大量的向量化場景驅動,不斷提升性能螞蟻專業算法團隊進行算法優化*圖中最上方 vsag 即為 OceanBase 的向量化引擎插件4.x 新特性：高可用Binlog服務OceanBaseBinlog Server 1Binlog Instance A（主）Binlog文件Binlog Instance B（備）Binlog Instance.Binlog Server 2Binlog Instance A（備）Binlog文件Binlog Instance B（主）Binlog Instance.OBProxyclogMetaDBBinlog Events非 Binlog

46、SQLBinlog SQL在最新的 4.x 版本中，OceanBase Binlog 服務開源了集群高可用特性，在對 MySQL Binlog 高度兼容的基礎上，顯著增強了服務的穩定性。Flink X OceanBase 數據遷移方案介紹遷移工具介紹：Flink Jdbc ConnectorFlink Jdbc Connector表結構查詢庫、表查詢單表數據讀取查詢下推單表數據寫入維表數據緩存為了更好地對接 Flink 生態，OceanBase 在 Flink Jdbc 連接器中實現了 OceanBase 的方言以及 Catalog（MySQL 和 Oracle 模式皆有）。Flink MyS

47、QL CDC（對接 OceanBase Binlog 服務）整庫同步表結構變更同步分庫分表合并Streaming ETL全增量一體化元數據千表入倉無鎖讀取斷點續傳并發讀取Flink OceanBase CDC（對接 OceanBase LogProxy 服務）整庫同步分庫分表合并Streaming ETL全增量一體化元數據千表入倉無鎖讀取斷點續傳遷移工具介紹：Flink CDC 2.x（Source）OceanBase 長期參與 Flink CDC 社區，目前的正式版本中，除了貢獻有一個 OceanBase CDC Source 連接器之外，OceanBase Binlog 服務已經將 Fli

48、nk MySQL CDC Source 連接器作為日?；貧w測試的主要工具之一?；?Flink CDC 2.x 的數據同步方案Flink CDCFlink JDBC 連接器Flink Kafka 連接器TiDBPolarDBRDS MySQLFlink Kafka 連接器+Flink JDBC 連接器優點：支持多種數據源、全增量一體化、Exactly Once缺點：多表同步需要借助 Kafka、無法利用 OceanBase 的分區表特性和旁路導入特性Flink 寫連接器優化：多表寫入RecordSerializationSchemaTableInfoRowDataFields,KeysFiel

49、d Values ArrayOceanBase JDBCDialectDML SQLOceanBase 開源團隊參考 Flink Jdbc Connector，獨立維護了一組 Flink 寫連接器，并實現了多種優化https:/ DataChangeRecord，支持 DataStream 模式下同時傳遞表結構和行數據，進而實現多表數據寫入通過引入 SchameChangeRecord，支持 DataStream 模式下傳遞 DDL 事件，進而實現 DDL 同步DataChangeRecordSchemaChangeRecordDDL SQL234計算 RowData 分區并發寫入單分區數據1

50、數據 buffer根據分區攢批Flink 寫連接器優化：分區表寫入優化對于分區表，按分區寫入數據，減少分布式事務，提高寫入效率經過內部測試和社區用戶使用驗證，優化后的寫入性能是普通 SQL 批量寫入的 3 倍以上Flink 寫連接器優化：接入旁路導入經測試，該連接器的數據寫入性能是 JDBC 連接器的 6 倍以上實踐性能提升情況旁路導入功能更適合大量歷史數據的遷移場景，因此我們為 OceanBase 單獨實現了一個旁路導入的 Flink 寫連接器，它使用 Flink 的 BATCH 運行模式，基于 OceanBase 的旁路導入 Java SDK 實現，通過 RPC 端口與 OceanBase

51、 進行交互，繞過了 SQL 層，極大的提高了數據寫入的效率。Flink 寫連接器優化：接入旁路導入應用程序Flink創建旁路導入客戶端directLoader directLoader.begin創建 Flink 任務SinkWriter 0directLoader.writeSinkWriter 1directLoader.writeSinkWriter.directLoader.writeFlink 任務執行結束directLmitOceanBaseSQL 層數據Flink CDC 3.0 開始新引入了可用于 YAML 作業的 Pipeline Source 和 Sink 連接器。如右圖所

52、示，配置一個 YAML 后，通過執行腳本即可提交一個端到端的同步任務。bash bin/flink-cdc.sh mysql-to-starrocks.yaml遷移工具介紹：Flink CDC 3.xFlink CDC 3.0端到端同步表結構變更同步整庫同步（Source+Sink）數據轉換DataChangeRecord、SchemaChangeRecordSnapshot Row、Log Event基于 Flink CDC 3.x 的數據同步方案MySQL CDC SourceOceanBase SinkFlinkFlink CDCCDCRouter、Transformer、Schema

53、EvolutionDataChangeEvent、SchemaChangeEventOceanBaseOceanBase PipelinePipeline SinkSinkOceanBase Binlog 服務完全適配 Flink MySQL CDC Source 連接器，因此完全可以將OceanBase Binlog 服務視作 MySQL 數據庫使用。OceanBase 基于自己維護的寫連接器，實現了 OceanBase 的 Sink Pipeline。目前正在社區的代碼評審流程中。遷移工具介紹：OceanBase遷移服務（OMS）HBaseOceanBaseOceanBase遷移服務表結構

54、遷移建立增量通道全量數據遷移增量數據遷移Flink消息隊列全量數據校驗優點：管控功能齊全白屏界面，易于操作數據校驗缺點：適配數據源類型比較少Flink 任務類型不夠通用，引入新數據源需要從頭開發打通 OMS 和 Flink 生態目前OceanBase正在基于自維護的 Flink 寫連接器開發命令行工具，期望實現對 Flink CDC 和 Flink JDBC 連接器的數據源接入。后續該工具也將集成到 OMS 當中，供社區用戶使用。Flink JDBCFlink CDCFlink Connector OceanBase寫連接器命令行工具OceanBase 數據遷移服務（OMS）TiDBPolar

55、DBRDS MySQLFlink X OceanBase 實時數據分析基于列存構建實時數據分析應用Flink CDCFlink JDBC數據應用實時寫入TP 數據庫OceanBase 數據庫基于列存引擎、并行執行引擎、向量化引擎，以及全新的優化器，具備擔任輕型數倉的能力，數倉場景可以全部使用 OceanBase 數據庫來簡化 ETL。列存表ODS列存表DWD列存表DWS列存表ADS維度表維表關聯多源合并指標計算復雜計算物化視圖示例：基于列存表構建實時 AP 應用MySQL源表OceanBase 列存表感興趣的同學可以訪問 OceanBase 博客網站，查看基于 Flink 的 OceanBas

56、e AP 實時分析 demo。https:/ DEMO 未來展望01Flink命令行工具02多模API03向量數據庫/AI完善 Flink 命令行工具，實現 OMS 與 Flink 生態的全面對接OceanBase 多模 API（HBase、Redis）全面適配 Flink 生態OceanBase 向量數據庫能力適配 Flink 生態THANK YOU謝 謝 觀 看Amoro+Flink CDC 數據融合入湖新體驗Apache Amoro PPMC&社區運營Apache Flink Contrubitor&社區志愿者陳政羽貨拉拉數據湖方向負責人CDC在貨拉拉應用Flink CDC 實時數據訂閱

57、入口數據入湖新體驗pipeline 入湖入湖優化入湖數據新鮮度提升04未來規劃CDC&數據湖Flink CDC 在貨拉拉應用CDC逐步成為集團實時數據訂閱入口Flink CDC 使用量數據量 TB+采集數據量穩步上升任務 50+建設標準化采集平臺采集 5000+表包含多個分庫分表落地場景公司內部目前的業務有小拉出行、貨拉拉、LaLaMove、跑腿等多業務線，同時還區分國內和海外多DC環境，整體業務數據量達到 TBPB 級別，目前已經接入的已經有實時看板、云臺、kepler、BI 報表、Monitor、交易2.0等業務，實現數據訂閱鏈路的“以舊煥新”，后續將持續推進老鏈路替換工作，完成最終平臺化

58、工程建設穩定性建設應用層直接提供給飛流任務和多方業務使用平臺適配能力上面我們對配置化、感知、協議、SDK等做了適配穩定性方面做了限流、HA、血緣、性能驗證等多方面測試和改進數據采集入湖場景離線計算增量計算流計算訪問要求高吞吐，低響應在延遲和吞吐獲取平衡低延遲，高響應訪問方式文件（行列）讀寫行列隨機讀寫記錄級別讀寫存儲周期長中長短（般7天）數據新鮮度低中高 埋點數據上報，數據具備時效性要求低，非結構化數據 CDC 數據入湖分析，數據時效性要求高，結構化數據 日志數據入湖存儲，需要間接性統計和分析，非結構化數據CDC入湖新體驗新特性、碎片文件、數據文件優化CDC 3.0 YAML With Pai

59、monTips：目前 iceberg 正在開展中，使用 iceberg flink connector v2 開發RouteSource/SinkTransformPipelineSchema Evolution數據入湖困痛點-文件碎片化嚴重Tips：Flink 每次進行Checkpoint時候，二階段提交把元數據和數據本身寫入到文件系統Paimon 會通過自動優化機制定期進行小文件自動合并，合并是黑盒操作數據入湖痛點-Schema 演進添加的列不會去讀現有列上的數據刪除一個列不會影響其他列更新一個列不會影響其他列的值更新列順序不會影響對應列的值Data SourceSchema Operat

60、orData SinkSchemaRegistryTableUpstreamSchemadb.table1id INT,TableEvolvedSchemadb.table1id INT,Tips：湖倉表模式演化的合理配置數據入湖痛點-數據質量管理Tips：隨機抽樣對比、整體對數對比BINLOG采集數據正確性轉換為CDC類型CDC類型轉換為湖倉識別類型寫入特殊類型處理（DateTime、TimeStampGEOMETRYCOLLECTION.）默認值設置數據入湖痛點-采集穩定性Tips：GH-OST/PG-OST 已經在最新版本CDC支持數據洪峰GCbinlog采集多線程解析提升速度DBA變更

61、工具支持支持GH-OST Schema change采集告警對采集DDL、DML變更預警下游作業基于 Amoro+CDC 湖倉融合一體架構存儲層計算層Flink自動調優流式入湖/讀取Amoro湖倉元數據優化作業管控COS埋點數據交易2.0數倉業務用戶畫像Data OSSHDFSS3.GPT速查AI+Flink CDC索引優化小文件合并Flink SQL.KeplerAmoro入湖優化針對iceberg碎片文件、數據文件優化湖倉管理系統 Apache AmoroAMORO 特點Amoro 利用 AMS 管理功能和自優化機制，為用戶提供了解決三方悖論的方案。與傳統的數據倉庫不同，Lakehouse

62、 表格被廣泛應用于各種數據管道、AI 和 BI 場景。衡量數據新鮮度對于數據開發人員、分析師和管理員至關重要，Amoro 通過在流計算中采用水印概念來解決這一挑戰，從而評估表格新鮮度。CDC 數據入湖困難點數據質量問題數據Schema change數據時效性問題讀放大和寫放大問題.CDC數據入湖優化 流式寫入文件碎片 優化資源管理 優化資源任務隔離 表格式讀寫平衡文件碎片優化過程（GC）數據優化階段Self-optimizing typeInput spaceOutput spaceInput file typesOutput file typesminorfragmentfragment/s

63、egmentinsert,eq-delete,pos-deleteinsert,pos-deletemajorfragment,segmentsegmentinsert,eq-delete,pos-deleteinsert,pos-deletefullfragment,segmentsegmentinsert,eq-delete,pos-deleteinsertTips：入湖時，可以根據數據新鮮度要求調整文件合并頻率Flink Mixed Format借助 Amoro Mixed Iceberg 有主鍵表，既支持了 ODS 層的 upsert，也支持增量消費 CDC 數據，繼續構建下游的表；尤

64、其是還可以通過 Log Store 將數據寫入 Kafka，給下游提供毫秒級延遲的數據，避免數據流經多個層級后延遲不斷疊加CDC實時入湖鏈路存儲層OSSOSSOSSOSSMySQLFlink CDC 實時 ETL 寫入PGCDC入湖優化效果對比12 Hours6 Hours1 Hours30 Mins050020001,5002,000優化前文件數優化后文件數基于 Flink 1.20 LTS、iceberg 1.6.X 實時寫入，Checkpoint 設置為 3 分鐘902201501627232104301612未來規劃社區規劃Amoro入湖Amoro支持多種數據一鍵入湖穩定性建設支持 F

65、link 1.20 LTS 版本Paimon 優化Amoro支持Paimon自動優化Mixed Format基于iceberg的實時入湖方案Iceberg 生態支持支持Pipeline寫入iceberg平臺化聯合Dinky+Amoro打造開源CDC入湖產品未來規劃數據湖CDCTHANK YOU關 注 公 眾 號 了 解 更 多 資 訊Flink CDC 3.0在醫保局多中心多庫多表場景應用實踐李斌松 數新智能-DataCyber技術負責人嘉賓簡介數新智能-DataCyber技術負責人李斌松10 多年大數據從業經驗參與或主導過阿里、同盾等公司大數據平臺架構設計與升級現負責數新智能(DataCyb

66、er)數據智能平臺研發項目背景解決方案增強解析項目成果項目背景及挑戰某省醫保局原有平臺架構國家平臺省級平臺地市1備份庫地市2備份庫地市3備份庫地市4備份庫地市5備份庫地市N備份庫地市1生產庫地市3生產庫地市4生產庫地市5生產庫地市N生產庫地市2生產庫T+1 定時拉取T+1 離線定時同步離線同步離線同步離線同步離線同步離線同步離線同步離線同步實時同步MPP 分析庫Hive 數倉現有問題l數據同步鏈路多l數據質量低l數據同步時效性差l數據同步維護成本高面臨挑戰多數據庫實例多數據中心多邏輯/物理表多目標 SinkFlink CDC 3.0 數據集成方案Flink CDC 實時入湖DWDODS關系數據

67、庫消息隊列DWSDWDDWSFlink CDC 實時入湖Flink 實時入湖Flink 實時鏈路Spark 離線鏈路技術選型Flink CDC 2.0l 單表同步l 不支持 Schema 變更l SQL 或者 API 定制Flink CDC 3.0l 分庫分表同步l 整庫多表同步l Schema 變更VS升級后數據平臺架構國家平臺省級平臺離線庫(數據湖Paimon）省級平臺實時庫(華為DWS、騰TDSQL/TBase等)Kafka各地市平臺數據集成數據開發數據質量數據對比T+1 定時拉取實時查詢 實時與離線同步鏈路合并 引入 Kafka 中轉多 Sink Paimon 實時入湖，提升時效性數新

68、智能CyberData運維中心Flink CDC 3.0Flink CDC 3.0阿里云RDS騰訊云RDS華為云RDS華三云RDSFlink CDC 3.0 能力增強解析數據集成整體鏈路MySQLKa-aPaimon Append Paimon UpsertDWS(PG)TBase(PG)TDSQL(MySQL)2135443.1 Kafka Sink 輸出表結構變更解決方案 Schema Event 序列化成 Json 字符串，直接輸出到 Kafka SchemaChangeEvent 沒有原始 DDL 語句，無法輸出Debezium JSON 格式數據?，F有 Kafka Sink 忽略 S

69、chema Event問題Kafka SinkTopicSchema/Data JSONKafka Source3.2 減少作業數解決方案 擴展 Flink CDC 的 Composer 模塊，支持配置多個 DataSource 多地市匯聚至同一 Topic，通過 Transform 增加擴展字段（areaCode等）一個任務消費多個 Topic，不同 Topic 寫入不同 Sink 表TopicSchema/Data JSONPaimon地市1地市2地市n多個分析數據庫一個邏輯表同步鏈路3.3 邏輯表 Schema 兼容性處理問題 在 Data Event 添加 Schema 信息 處理數據

70、時直接使用當前 Schema 信息解決方案地市同一業務表結構有微小差別，如字段類型、字段個數、字段順序等不一致3.4 支持邏輯刪除支持邏輯刪除1.支持物理刪除和邏輯刪除2.記錄刪除時間3.5 臟數據處理Transform 和 Sink 處理失敗，失敗 Event 存儲 Kafka1.任務 ID 和名稱2.MySQL 數據源信息3.Sink 信息4.失敗原因5.失敗時間Flink CDC 3.0 能力增強總結與社區官方深度合作，持續對 Flink CDC 3.0 優化升級l Pipeline 支持多 Sourcel DataEvent Schema 差異兼容性支持（字段名稱、字段個數、字段順序等

71、）l 新增 Kafka Source，增強 Kafka Sinkl 新增 MySQL、PostgresSQL Sinkl Transform 和 Sink 支持臟數據轉存l 所有 Sink 支持邏輯刪除Flink+Paimon 實時湖倉一體平臺資源調度數據計算數據開發治理平臺數據集成數據開發作業調度數據資產數據安全數據服務離線計算服務實時計算服務流批一體引擎（Flink）分布式資源調度(Yarn)統一元數據（Hive Metastore）數據存儲分布式文件系統（HDFS）+數據湖（Paimon）數據集成數據集成（Flink CDC 3.0）全量同步全增量同步實時同步離線同步整庫同步分庫分表同步

72、數據源異構數據源（關系型數據庫，消息隊列，文件系統等）實時任務開發：MySQL Kafka實時任務開發：Kafka Paimon實時任務監控項目成果實施效果700+數據庫實例15+數據中心500+邏輯表15000+物理表350+個任務降低系統維護成本實現6套系統合并提升數據時效性T+1 分鐘提升資源利用率Flink 集群計算資源減少90%以上提升數據治理水平支持數據比對能力支持臟數據處理能力達成效益THANK YOU謝 謝 觀 看Flink+StarRocks高并發實時導入優化李鵬飛鏡舟科技資深數據庫研發工程師StarRocks Active Contributor背景技術挑戰解決方案未來規劃

73、背景StarRocks極速統一湖倉新范式Flink+StarRocks構建實時數倉StarRocksFlink增量計算(UPSERT+DELETE)寬表Join數據亂序Exactly-OnceDDL/整庫同步主鍵模型(Delete Vector)部分列更新條件更新Transaction Stream LoadFast Schema EvolutionConnector適配Flink CDC 3.x高并發實時導入需求典型場景：網絡安全、電信運營商Flink Pipeline特點并發高，Transform需要高并發滿足CPU消耗，同時Transform和Sink chain到一起避免Shuffle

74、開銷一個Pipeline有多個表并且數量動態變化，無法對單表進行并發調優秒級實時，導入頻率高，從上游消費到StarRocks數據可見保證秒級技術挑戰StarRocks系統架構存算一體(Shared-nothing)存算分離(Shared-data)數據導入鏈路Sink每個并發內存攢批后通過一次 Stream Load 導入 StarRockssink.buffer-flush.max-bytes 控制內存使用sink.buffer-flush.interval-ms 控制導入延遲Sink使用HTTP Client發送 Stream Load 請求等價于如下curl命令Stream Load流程

75、存算一體和存算分離基本流程一樣，存算一體還需要考慮副本同步，以存算分離為例說明Stream Load 提供事務，保證ACIDTwo-Phase Commit：協調多個Tablet之間一致性，導入過程需要與 FE Leader 進行多次交互MVCC：每個Tablet產生一個數據版本，將數據持久化成多個文件，后續通過compaction合并多版本提高查詢性能Stream Load流程存算一體和存算分離基本流程一樣，存算一體還需要考慮副本同步，以存算分離為例說明Stream Load 提供事務，保證ACIDTwo-Phase Commit：協調多個Tablet之間一致性，導入過程需要與 FE Lea

76、der 進行多次交互MVCC：每個Tablet產生一個數據版本，將數據持久化成多個文件，后續通過compaction合并多版本提高查詢性能FE Leader容易成為瓶頸Compaction資源消耗增加I/O延遲增加IOPS達到限制后不易Scale高并發實時場景挑戰事務數版本數小文件數IOPS影響查詢性能TPS 表數*并發數*單并發導入頻率IOPS TPS*Tablet數*3(meta/data files)Obejct Store成本增加實際例子300表*300并發*1s=90K TPS90K TPS*5 Tablets*3=1.35M IOPS解決方案優化思路TPS 表數*并發數*單并發導入

77、頻率業務需求實時并發間不需要嚴格ACID，保證at-least-once/exactly-once即可，并發間的導入可以合并到一個事務300表*300并發*1s=90K TPS90K TPS*5 Tablets*3=1.35M IOPS300表*1s=300 TPS300 TPS*5 Tablets*3=4.5K IOPSFlink側合并?基于Flink Sink Coordinator協調各個并發寫入一個事務使用復雜度高，不易擴展到其它場景Beam on Flink無法使用 Sink Coordinator API非Flink場景可能沒有coordinator，比如用戶自己封裝的導入工具St

78、arRocks側合并(Merge Commit)按照 merge_commit_interval_ms 切分時間窗口，每個窗口內的導入請求合并到一個事務每個窗口部署一個分布式Plan，可以并行處理多個Coordinator CN的導入數據時間窗口到達后Plan自動結束，提交當前事務，下個時間窗口開啟新的事務Coordinator CN分配每個表通過 merge_commit_parallel 設置分布式Plan的并行度，即Coordinator CN的數量，并行度主要取決于導入負載根據集群當前可用CN以及并行度設置，為每個表分配 Coordinator CN 集合，將導入請求和Plan分發到該

79、集合，從而Plan可以讀取導入數據，進行合并多表場景下，所有表的并行度在CN上均勻分布，盡量保證CN之間負載均衡，表的數量動態變化時需要重新調度進行均衡根據CN狀態動態調整，比如節點宕機、Scale in/out 數據傳輸鏈路優化HTTP傳輸鏈路問題HTTP Server不能同時支持連接復用+異步為了避免請求之間相互阻塞，每個請求使用單獨TCP連接TCP并發連接數過多帶來一些開銷和問題，比如TCP Listen Queue Overflow數據傳輸特點峰值并發導入請求高，約為表數*并發數，比如checkpoint觸發所有數據同時flush，300表*300并發=90K導入請求高頻高并發下每個導

80、入請求數據量少，導入請求可以復用TCP連接進行數據傳輸數據傳輸鏈路優化使用 Apache bRPC 替換 HTTP 進行數據傳輸，支持連接復用+異步，可以減少TCP連接數，同時提高并發處理能力BRPC Server 增加 stream_load 接口Connector 引入開源框架 brpc-java 與server通信效果驗證一個Pipeline 300表,300并發1s寫入頻率，checkpoint interval 10s總流量90K events/s，表流量分布符合二八法則效果驗證一個Pipeline 300表,300并發1s寫入頻率，checkpoint interval 10s總流

81、量90K events/s，表流量分布符合二八法則未來規劃 StarRock 3.4 發布 支持 exactly-once 支持 INSERT INTO VALUES 集成其它生態組件THANK YOU謝 謝 觀 看Flink CDC+Hologres高性能數據同步優化實踐胡一博阿里云高級技術專家Hologres介紹實時數倉實時數倉OLAPOLAP分析分析數據服務數據服務ServingServing湖倉數據湖倉數據交互式分析交互式分析 百萬RPS實時寫入，寫入即可查 高性能實時整行更新和局部更新 高QPS、低延遲OLAP簡單查詢 復雜查詢：TPC-H 30TB世界性能第一 列存 百萬QPS K

82、V點查，高可用 達摩院Proxima向量檢索 全SQL表達 行存、行列共存 讀寫分離 OLAP分析和數據服務強隔離 對離線數倉MaxCompute、數據湖進行秒級交互式查詢 無需數據搬遷、元數據自動發現 百萬行每秒極速數據同步 內外表聯邦分析分析服務一體化（Hybrid Serving/Analytics Processing，HSAP）Hologres連接器實時寫入與更新寬表MergeDDL變更同步整庫同步全量讀取Binlog讀取CDC讀取全增量一體化讀取實時計算FlinkHologres維表Hologres CatalogHologres結果表Hologres源表維表查詢元數據查詢&更新維

83、表百萬RPS查詢維表實時可更新場景：離線維表、實時特征強大的實時寫入能力與整行更新能力，完美匹配Flink寬表局部更新，簡化多流Join元數據服務整庫同步Schema Evolution流批一體CDC全鏈路實時觸發Hologres實時寫入原理pkpklsnlsn110240pkpklsnlsndatadata110value1240value2Pk index行存（LSM）數據列存shardpkpklsnlsnXXXXpkpklsnlsndatadataXXXXXXshardhash(pk)Hologres實時寫入原理pkpklsnlsn110240pkpklsnlsndatadata110v

84、alue1240value2Pk index行存（LSM）數據列存shardpkpklsnlsn141240pkpklsnlsndatadata110value1240value2shardINSERT INTO table_nameVALUES(1,value3)ON CONFLICT(pk)DO UPDATE SETdata=excluded.data;pkpklsnlsndatadata141value3file1file1memtabledelete mapfilefilebitmapbitmapfile101.沖突檢測2.標記刪除3.插入數據Hologres binlog消費原理pk

85、pklsnlsn110240pkpklsnlsndatadata110value1240value2Pk index行存（LSM）數據列存shardpkpklsnlsn141240pkpklsnlsndatadata110value1240value2shardINSERT INTO table_nameVALUES(1,value3)ON CONFLICT(pk)DO UPDATE SETdata=excluded.data;pkpklsnlsndatadata141value3file1file1memtabledelete mapfilefilebitmapbitmapfile10bin

86、loglsnlsnpkpkdatadatatypetype401value1-U411value2+U1.沖突檢測2.標記刪除3.插入數據4.插入binlog寫入優化更短的數據可見性更高的并發度支持亂序場景數據保序更低的資源消耗消費優化更高的離線場景吞吐分區表支持增量消費未來展望寫入優化寫入優化連接器寫入模型Buffer queueBuffer queueBuffer queueJdbc connectionJdbc connectionTMBuffer queueBuffer queueBuffer queueJdbc connectionJdbc connectionTMDATADATAI

87、s queue full?INSERT INTO table_name VALUES(?,?,?),(?,?,?)ONCONFLICT(pk)DO UPDATE.DELETE FROM talbe_name WHERE PK=?OR PK=?hologres寫入優化更短的數據可見性Buffer queueBuffer queueBuffer queueJdbc connectionJdbc connectionTMDATABuffer queueBuffer queueBuffer queueJdbc connectionJdbc connectionTMDATAIs queue full?I

88、s connection idle?aggressive.enabled:true配置寫入優化更高的并發度支持Buffer queueBuffer queueBuffer queueJdbc connectionJdbc connectionTMDATAjdbc:postgresql:/HOST:PORT/DB?options=type=fixedhologresgatewayfrontendfixed frontendconnType=msdkMode:jdbc_fixed配置寫入優化亂序場景數據保序INSERT INTO table_name VALUES(?,?,?),(?,?,?)ON

89、 CONFLICT(pk)DO UPDATE.WHERE excluded.update_time table_name.update_time;pkpklsnlsnupdate_timeupdate_time11010:00:0024011:00:00pkpklsnlsndatadataupdate_timeupdate_time110value110:00:00240value211:00:00shardfile1減少反查IOcheck-and-put.column:update_timecheck-and-put.operator:GREATER配置寫入優化更低的資源消耗INSERT I

90、NTO table_name VALUES(?,?,?),(?,?,?)ONCONFLICT(pk)DO UPDATE.DELETE FROM talbe_name WHERE PK=?OR PK=?sqlclientserversql問題：攢批引起taskmanager內存需求高；單個連接沒有跑滿，需要flink更高并發COPY table_name FROM STDINWITH(STREAM_MODE true,ON_CONFLICT update);begin copyclientserverrecordend copy優點：無需等待返回，網絡損耗幾乎抹平的情況下批可以非常小，節省內存；

91、單個連接的吞吐大約能提升8倍sdkMode:jdbc_copy配置寫入優化更低的資源消耗SourcesinktasktasktasktaskrepartitionCOPY table_name FROM STDINWITH(ON_CONFLICT update);準實時（分鐘級）Hologres側cpu下降約70%reshuffle-by-holo-distribution-key.enabled:true配置寫入優化寫入模式選擇數據可見性reshuffle-by-holo-distribution-key.enabled:true是否包含回撤消息準實時實時sdkMode:jdbc_copys

92、dkMode:jdbc_fixed是否消費優化消費優化更高的離線場景吞吐物理文件arrowPg csvPg binary RowDataTMFEBESTORAGESELECT物理文件arrowarrowRowDataTMFEBE(serverless)STORAGECOPY優點:連接利用率更高的協議；節約一次格式轉換；減少對線上業務的影響消費優化分區表支持增量消費消費模型：每張子表每個shard持有一個單調遞增的lsn，占用一個連接-常量分區表，例如城市分區-時間分區表，例如月分區、日分區、小時分區fixed frontend支持binlog消費，連接數無硬性上限時間2024-10-01202

93、4-10-022024-10-03啟動位點2024-10-02=2024-10-02 00:00:00,2024-10-03 00:00:00)min max未來展望未來展望全增量消費時保證全量和增量階段lsn嚴格對齊精簡寫入模式-INSERT/DELETE-COPY -不支持分區表 -不支持回撤消息支持更多種類Schema Evolution-支持add column not null-支持alter column類型40+客戶案例合集：Hologres一體化實時湖倉平臺阿里云上客戶案例阿里巴巴集團案例互聯網平臺小紅書曹操出行錢大媽新氧醫美諾亞財富來電科技輕松籌玩物得志游戲娛樂37手游樂元素

94、TapTap心動玩吧軟件服務金蝶管易云萬里?？腿缭凭鬯稄V告營銷小邁科技飛書深諾物流交通遞四方運滿滿易倉快狗打車在線教育好未來新東方實時數倉CCO淘菜菜Lazada阿里媽媽菜鳥Aliexpress零售通交互式分析CCO菜鳥天貓國際淘特網絡部流量分析友盟Lazada淘天實時大屏飛豬淘天搜索推薦淘寶搜索搜推實時數倉搜推多維分析淘寶訂閱AI淘天掃碼查看（Hologres官網可查）免費試用5000CU時 100GB存儲THANK YOU謝 謝 觀 看Dinky 和 FlinkCDC 在實時整庫同步的探索之路The exploration process of Dinky and FlinkCDC in

95、 real-time full database synchronization亓文凱Dinky Maintainer,FlinkCDC Contributor起源探索未來起源數據集成的背景分析庫業務庫三點要求數據一致性鏈路穩定性數據時效性傳統數據集成方案分析庫業務庫DataX/Sqoop快照同步Debezium/Canal增量同步快照表增量表合并表全量增量割裂鏈路長組件多數據時效性低FlinkCDC 數據集成方案業務庫FlinkCDC實時一致性快照數據快照變更日志增量快照算法鏈路短組件少數據時效性高數據集成面臨的挑戰整庫同步業務庫FlinkCDCFlinkCDCFlinkCDCTable 1

96、Table 2Table NSink Table 1Sink Table 2Sink Table N占用大量連接數Binlog 重復讀取開發運維大量任務數據集成面臨的挑戰模式演變業務庫FlinkCDC分析庫表結構變更未同步新列數據未同步CDC 可能出現異常idnameage1Jack20idname1Jack發生異常用戶渴望端到端數據集成業務庫分析庫全增量自動切換元數據自動發現表結構變更自動同步整庫同步只用一個鏈接一行SQL部署整個作業探索全增量自動切換FlinkCDC 增量快照算法對業務庫影響降低只占用一個連接數直觀表達任務需求一句 SQL 或一個 Yaml 腳本元數據自動發現根據元數據構建

97、任務細節及建表表結構自動演變阻塞數據保證語義正確自定義處理邏輯分庫分表、表映射及數據轉換整庫同步的探索之路Dinky CDCSourceFlink CDCPipelineDinky 的介紹實時計算 IDE提供 FlinkSQL、FlinkJar、FlinkUDF 作業開發，擴展了很多類 FlinkSQL 語法來增強 Flink。實時任務運維鏈接 Flink 集群，對 Flink 作業進行實時監控及異常報警，對作業狀態及版本進行管理。Dinky 是一個以 Apache Flink 為內核構建的開源實時計算平臺，連接數據湖倉等眾多開源框架，致力于批流一體和湖倉一體的建設實踐，助力實時計算高效應用。

98、Dinky 基于 FlinkCDC 的整庫同步探索業務庫分析庫FlinkCDC+DataStreamidnameage1Jack20idnameage1Jack20全增量自動切換元數據自動發現結構變更自動同步只用一個庫鏈接支持讀FlinkCDC源一句 SQL 部署作業支持寫各種數據源支持所有部署模式Dinky CDC Source 整庫同步數據鏈路Dinky CDC Source 基于 Flink CDC 在分庫分表合并場景下：Data SourcePartitionRouteOutput TagFlat MapData SinkDebezium Json按表名+主鍵分區DataChangeE

99、vent按表名合并后分發table1table1table2table2Dinky CDC Source 整庫同步數據鏈路Dinky CDC Source 基于 Flink CDC 在模式演變、分庫分表合并場景下：Data SourcePartitionRouteOutput TagTo StringDoris SinkDebezium Json按表名+主鍵分區DataChangeEvent按表名合并后分發SchemaChangeEventtable1table2table1table2Dinky CDC Source 的局限性大量算子節點作業龐大不支持自定義轉換只能進行寬容數據類型轉換模式演

100、變受限模式演變由下游 Sink 連接器自行實現Dinky CDC Source 整庫同步的數據轉換探索Dinky CDC Source 基于 Flink CDC 在整庫同步中進行數據轉換的場景探索：DataChangeEventFlat MapTemporary ViewFlink CTASData SinkData SourceFlink Row使用 CTAS 中的Select 設置轉換轉為 Table Api已省略前驅環節構建大量算子節點不支持模式演變Flink CDC YAML 定義的 Pipeline 作業FlinkCDC 通過一個 Yaml 腳本定義整庫同步任務，支持模式演變、分庫分

101、表合并、數據轉換等能力。FlinkCDC YAML 核心架構部署方式YarnStandaloneKubernetes運行環境Flink RuntimeFlink CDCDataSource OperatorDataSink OperatorSchema RegistryRouterTransformerFlink CDC ComposerMySQL SourceElastic Search SinkKafka SinkPaimon SinkStarRocks SinkDoris SinkFlink CDC CliYaml流處理模式演變應用場景整庫同步分庫分表同步ETLFlinkCDC Pipe

102、line 整庫同步數據鏈路FlinkCDC Pipeline 在模式演變、數據轉換、分庫分表合并場景下：Data SourcePreTransformPostTransfomSchema OperatorPartitionData Sink原始 Schema裁剪 Schema投影 Schema合并 Schema刪除投影無關列條件過濾數據生成計算列合并分表結構按分區分發數據模式演變阻塞數據SchemaChangeEventDataChangeEventFlink CDC Pipeline 的 Transformtransform:-source-table:test.userinfoprojec

103、tion:id,age_year(birthday)agefilter:age=18 and age 60primary-keys:idpipeline:user-defined-function:-name:addoneclasspath:com.cdc.udf.examples.java.AgeYearFunctionClass PreTransformPostTransfom.Data SinkData SourceSchema:id INT,name STRING,birthday STRINGData:1,A,2006-01-012,B,2007-01-013,C,1946-01-0

104、11,A,2005-01-01Schema:id INT,birthday STRING,PRIMARY KEY(id)Data:1,2006-01-012,2007-01-013,1946-01-011,2005-01-01Schema:id INT,age INT,PRIMARY KEY(id)Data:1,181,19Schema:id INT,age INT,PRIMARY KEY(id)Data:1,19鍵定義列投影條件過濾使用 UDFTransform 的實現細節PreTransformPostTransfomSchemaChangeEventDataChangeEventEven

105、t推導所引用的列消除未引用列緩存裁剪后Schema裁剪數據列主鍵、分區鍵定義裁剪后 Schema 演變SchemaChangeEventDataChangeEventEvent推導計算列添加計算列緩存投影后Schema條件過濾計算投影投影后 Schema 演變Transform 的表達式計算技術實現的兩大框架：Janino:輕巧的 Java 動態編譯器，在 Apache Flink 中使用。Apache Calcite:一個靈活的、可擴展的 SQL 解析和優化框架，在 Apache Flink 中使用。Calcite注冊表元數據解析語法樹推導邏輯結構獲取最終表結構注冊內置函數Transform

106、Def投影規則用戶自定義函數過濾規則運行時 Schema運行時 Data內置函數Janino導入內置函數映射腳本邏輯樹生成表達式語句緩存表達式遍歷 SQL 語法樹計算時調用表達式不等于 Flink SQL模式演變的機制與策略SchemaChangeEventDataChangeEventFlushEventData SourceSchema OperatorPartitionData SinkSchemaRegistry1.收到Schema變更事件2.阻塞上游新事件3.向 Registry確認變更4.向下游廣播Flush 事件5.Flush操作完成后通報Registry6.向外部數據庫應用變更

107、7.通報 Schema變更結束8.停止阻塞，繼續處理數據2.其他并行數據源也阻塞全局演變策略：Lenient(Default)：調整表結構為最寬類型，不會丟失任何信息。Ignore：忽略所有表結構變更。Try Evolve：嘗試進行表結構變更，失敗則對數據進行轉換。Evolve：嘗試進行表結構變更，失敗則拋出異常。Exception：拒絕任何結構變更，一旦發生拋出異常。細粒度策略：include.schema.changes:需要同步的表結構變更類型。exclude.schema.changes:不需要同步的表結構變更類型。其他方面比較分庫分表數據結構輸入輸出Flink CDC 的高性能數據結

108、構Dinky CDC Source攜帶 SchemaFlink CDC Pipeline無 Schema+二進制id:INTname:STRING1Jackid:INTname:STRING2Bobid:INTname:STRING.id:INTname:STRING1Jack2Bob.Flink SQL ChangelogInsert(+I)Flink CDC EventDataInsertafterUpdate Before(-U)Update After(+U)Delete(-D)UpdatebeforeafterDeletebeforeFlink CDC 的分庫分表寬容機制id:INT

109、name:STRING1Jackid:INTname:STRINGaddress:STRING2Bob上海市id:INTname:STRINGaddress:STRING1Jark2Bob上海市Flink CDC Pipelineid:INTtotal:INT199id:INTtotal:BIGINT212345678912345id:INTtotal:BIGINT199212345678912345Flink CDC Pipelineid:INTname:STRING1Jarkid:INT2id:INTname:STRING1Jark2Flink CDC Pipeline新增列修改列刪除列分

110、表一分表二目標表Source 與 SinkDinky CDC SourceData SourceFlink CDC PipelineData SourceData SinkData SinkMySQLDorisStarRocksPaimonKafkaElasticsearchMySQLOraclePostgreSqlServer所有 Flink CDC SourceDorisKafkaHive所有 Flink SQL SinkPaimonHudi兩者差異總結Dinky CDC SourceFlink CDC Pipeline作業定義類 SQL 語句Yaml 腳本提交方式界面化提交Shell 命

111、令模式演變由 Sink 實現完整的演變機制及策略分庫分表支持同結構合并支持不同結構寬容合并數據轉換只有類型寬容轉換自定義過濾和投影Source 支持支持所有 FlinkCDC SourceMySQLSink 支持支持所有 FlinkSQL SinkDoris,StarRocks等共五個數據結構Flink Row高性能二進制結構算子拓撲節點大量非常少部署模式全部不支持 Yarn資源消耗較高較低在 Dinky 中使用 Flink CDC PipelineDinky 引入 Flink CDC Pipeline 所需要的 Jar 包后，通過編寫類SQL 的 EXECUTE PIPELINE 語句來定義

112、原先 Yaml 腳本，然后可以增強 Flink CDC Pipeline 任務，支持全模式部署，享受實時計算 IDE 的開發與運維能力。讓 Flink CDC Pipeline 縱享絲滑。未來Dinky 及 Flink CDC 規劃Dinky完善 CDC Source 的異常提示添加 CDC Source 轉 PipelineCDC Source 支持設定細粒度策略CDC Source 支持 Transform 探索實時湖倉治理尋找 Apache 孵化器導師Flink CDC支持離線場景Transform 支持調用 AI 模型擴展 Pipeline 上下游連接器支持更豐富的表結構變更類型支持更豐富的數據類型支持 Flink 1.20THANK YOU謝 謝 觀 看