FFA2024分論壇-數(shù)據(jù)集成 合輯

FlinkCDC是基于數(shù)據(jù)庫的日志CDC（ChangeDataCapture）技術(shù)，流式數(shù)據(jù)集成框架，配合Flink優(yōu)秀的管道能力和豐富的上下游生態(tài)，F(xiàn)linkCDC可以高效實(shí)現(xiàn)海量數(shù)據(jù)的實(shí)時集成。全量數(shù)據(jù)實(shí)時,一致性Paimon增量數(shù)據(jù)Paimon增量數(shù)據(jù)…Iceberg FlinkSQLAPIFlinkDataStreamAPIaggregatejoinJOIN鏈路組件多數(shù)據(jù)新鮮度數(shù)據(jù)一致性研發(fā)技術(shù)棧鏈路組件多數(shù)據(jù)新鮮度數(shù)據(jù)一致性研發(fā)技術(shù)?！说蕉俗鳂I(yè)不丟不重亞秒級延遲端到端作業(yè)不丟不重亞秒級延遲…Iceberg FlinkSQLAPIFlinkDataStreamAPIaggregatejoinJOIN全增量一體化2020/072021/082022/112023/102024/012024/09tabletable一Task2Task3Task1FlinkCDC3.x：端到端實(shí)時數(shù)據(jù)集成框架Analytics/BIDatabaseDataLakeDataWarehouse息StarRocksStarRocksPaimon$>flink-cdc$>flink-cdc.shmysql-to-starrocks.yaml mysql-to-starrocks.yamlDebeziumDebeziumInsertAfterCreateTableEventInsertAfterDataChangeEventInsertAfterCreateTableEventInsertAfterDataChangeEventAddColumnEventTruncateTableEvent…SchemaChangeEventAddColumnEventTruncateTableEvent…SchemaChangeEventStreamRecordStreamRecord1U13311U1331U133DataChangeEventSchemaChangeEventSchemaChangeEventBinaryData+SchemalessIDLEWAITINGIDLEWAITINGFINISHEDAPPLYINGFINISHEDAPPLYINGSchemaregistryappliesschemachangeSchemaChangeEventDataChangeEventSchemaChangeEventDataChangeEventFlushEventSchemaregistryacceptsschemachange,andrejectsfollowingrequestsFlushEventSchemaregistryacceptsschemachange,andrejectsfollowingrequestsSchemaoperatorregistersschemachangethenwaitforresponse(holdupstream).BlocksifSchemaRegistryisbusySinknoti?es?ushcompleteSchemaregistrycon?rmsschemaevolutioncompletes,readyfornextrequestSchemaoperatorbroadcastsFlushEvent,requestsregistryagaintowaitfor?ushcompleteSchemaoperatorreceivesSchemaChangeEventSchemaoperatorreleasesupstreamtherschemaoperatormustwaituntilotherschemachangesare SchemaOperatorDataSource SchemaOperatorDataSource PostPartitionerDataSinkPaimonPaimonStarRocks………l… 業(yè)務(wù)場景：?RAG(檢索增強(qiáng)生成)00 0 0 支持AIModel（已有PR）支持Batchpipeline對接更多上下游，如：Iceberg，ClickHouse支持更多類型，包括schemachange類型和eventchange類型擁有獨(dú)立的文檔網(wǎng)站:/?ink/?ink-cdc-docs-stableTHANKYOU全行擁有273家分支機(jī)構(gòu)網(wǎng)點(diǎn)覆蓋長三角、珠三角環(huán)渤海灣等發(fā)達(dá)經(jīng)濟(jì)圈省內(nèi)地市實(shí)現(xiàn)網(wǎng)點(diǎn)全覆蓋杭州銀行成立于1996年9月，是一家典型的城市商業(yè)銀行，堅持服務(wù)區(qū)域經(jīng)濟(jì)、中小企業(yè)和城鄉(xiāng)居民的市場定位，致力于為客戶提供專業(yè)、便捷、親和、全面的金融服務(wù)2016年10月27日，在上海證券交易所成功掛牌上市，截至2023年底，總資產(chǎn)18413.42億元，2023年營業(yè)收入350.16億元，利潤收入162.82億元。l在英國《銀行家》雜志2023年公布的全球銀行1000l2023年《財富》中國500強(qiáng)企業(yè)位列第380位l連續(xù)蟬聯(lián)“十佳城商行”l最具競爭力中小銀行l(wèi)最佳普惠金融成效獎l浙江省融資暢通工程“突出貢獻(xiàn)獎”l最佳科技金融服務(wù)城商行l(wèi)支持浙江省經(jīng)濟(jì)社會發(fā)展優(yōu)秀單位一等獎40l獲評“杭州市模范集體”DBlink初探FlinkCDC20242023201720072024202320172007DataX增強(qiáng)FlinkCDC需要更快的同步數(shù)據(jù)，提高數(shù)….…..….…..理 高可用高可用yarnyarn….拿批流拿Case1:基于watermark自動生成Tag….拿批流拿….….↓↓Catalog/Database/TableCatalogBaseCatalog/Database/TableCatalogBase否數(shù)據(jù)開發(fā)方否是采集平臺方否數(shù)據(jù)開發(fā)方否是采集平臺方源系統(tǒng)方_」_」!!JavaClientJavaClient!!!THANKYOU 完全自主研發(fā)的數(shù)據(jù)庫，才能做到核心替代的真正落地完全自主研發(fā)的數(shù)據(jù)庫，才能做到核心替代的真正落地—原生多租戶架構(gòu)—自研一體化架構(gòu)兼容經(jīng)典模式，實(shí)現(xiàn)單機(jī)和分布式、TP和AP的融合—原生多租戶架構(gòu)—自研一體化架構(gòu)兼容經(jīng)典模式，實(shí)現(xiàn)單機(jī)和分布式、TP和AP的融合—集中式/分布式一體化架構(gòu)——集中式/分布式一體化架構(gòu)——基于Paxos+數(shù)據(jù)同步的靈活的容災(zāi)架構(gòu)——基于Paxos+數(shù)據(jù)同步的靈活的容災(zāi)架構(gòu)—★低成本—靈活的部署模式——靈活的部署模式—★多租戶用純列存表行列冗余表純列存表createtablet2(c1createtablet2(c1varchar(1),c2varchar(1))withcolumngroup(eachcolumn);c1varchar(1),c2varchar(1))withcolumngroup(allcolumns,eachcolumn);語法說明語法說明l基于資源組的資源隔離方案l基于資源組的資源隔離方案lAP業(yè)務(wù):多個AP業(yè)務(wù)的處理優(yōu)先級不同基于資源組的資源隔離方案基于資源組的資源隔離方案…..…..…..…..…..…..??快速執(zhí)行路徑?優(yōu)化寫入放大問題?表鎖阻塞寫入insert/*+appendenable_parallel_dmlparallel(3)*/intot1select……;loaddata/*+direct(true,1024)parallel(16)*/in?le……;Tablet’04xlarge(16c32G)圖圖…….★不僅僅是KV★不僅僅是KV★通用的使用場景★易用的分布式存儲★易用的分布式存儲基礎(chǔ)類型HBase基礎(chǔ)類型HBase融合查詢并行執(zhí)行融合查詢并行執(zhí)行多值索引全文索引多值索引全文索引多分區(qū)水平擴(kuò)展高效存儲讀寫分離多分區(qū)水平擴(kuò)展高效存儲讀寫分離?提供?提供SQL客戶端和PythonSDK兩種使??式建表語句，向量作為一個單獨(dú)的列createtablet1(c1vector(3),c2int,c3?oat,primarykey(c2));數(shù)據(jù)導(dǎo)入，使用標(biāo)準(zhǔn)SQL語法導(dǎo)入insertintot1values('[7576.42,467.23,2913.762]',3,54.6);使用DDL進(jìn)行SQL語法，可以指定索引類型和距離算法CREATEINDEXvidx_c1_t1ont1(c1l2)usinghnsw;帶有向量計算的SQL，優(yōu)化器會自動使用向量索引OceanBase顯著增強(qiáng)了服務(wù)的穩(wěn)定性。OceanBase的分區(qū)表特性和旁路導(dǎo)入特性的分區(qū)表特性和旁路導(dǎo)入特性?對于分區(qū)表，按分區(qū)寫入數(shù)據(jù)，減少分布式事務(wù)，提高寫入效率?經(jīng)過內(nèi)部測試和社區(qū)用戶使用驗(yàn)證，優(yōu)化后的寫入性能是普通SQL批量寫入的3倍以上數(shù)據(jù)數(shù)據(jù)bu?er根據(jù)分區(qū)攢批計算RowData分區(qū)并發(fā)寫入單分區(qū)數(shù)據(jù)OceanBaseOceanBasedirectLoader.begin!bashbashbin/?ink-cdc.shmysql-to-starrocks.yamlSchemaChangeEventOceanBasePipelineSink?管控功能齊全?適配數(shù)據(jù)源類型比較少數(shù)據(jù)源需要從頭開發(fā)OceanBase?管控功能齊全?適配數(shù)據(jù)源類型比較少數(shù)據(jù)源需要從頭開發(fā)OceanBase目前OceanBase正在基于自維護(hù)的Flink寫連接器開發(fā)命令行工具，期望實(shí)現(xiàn)對FlinkCDC和FlinkJDBC連接器的數(shù)據(jù)源接入。后續(xù)該工具也將集成到OMS當(dāng)中，供社區(qū)用戶使用。OceanBase數(shù)據(jù)庫基于列存引擎、并行執(zhí)行引擎、向量化引擎，以及全新的優(yōu)化器，具備擔(dān)任輕型數(shù)倉的能力，數(shù)倉場景可以全部使用OceanBase數(shù)據(jù)庫來簡化ETL。OceanBase列存表OceanBase列存表/blog/12393391987 完善Flink命令行工具，實(shí)現(xiàn) OceanBase多模API（HBase、Redis）全面適配 OceanBase向量數(shù)據(jù)庫能力THANKYOUApacheAmoroPPMC&社區(qū)運(yùn)營ApacheFlinkContrubitor&社區(qū)志愿者用配衡長低中高 ............化機(jī)制，為用戶提供了解決三方悖論的方案。與傳統(tǒng)的數(shù)據(jù)倉庫不同,Lakehouse表格被廣泛應(yīng)用于分析師和管理員至關(guān)重要，Amoro通過在流計算中采用水印概念來解決這一挑戰(zhàn)，從而評估表格新鮮度。?數(shù)據(jù)Schemachange題?....fragmentfragment/segminsert,eq-delete,pos-deleteinsert,pos-deletefragment,segmentsegmentinsert,eq-delete,pos-deleteinsert,pos-deletefragment,segmentsegmentinsert,eq-delete,pos-deleteinsertIceberg有主鍵表,既支持了ODS層,繼續(xù)構(gòu)建下游的表；尤其是還可以數(shù)據(jù)寫入Kafka，給下游提供毫秒級延遲的數(shù)據(jù)，避免數(shù)據(jù)流經(jīng)多個層級后延遲不斷疊加0430THANKYOU ODSODSKafkaKafkaKa-aTBase(Ka-aTBase(PG)23445TopicKafkaSourceTopicKafkaSource地市1Topic多個分析數(shù)據(jù)庫地市1Topic多個分析數(shù)據(jù)庫Paimon地市2地市nTHANKYOU?增量計算(UPSERT+DELETE)典型場景：網(wǎng)絡(luò)安全、電信運(yùn)營商FlinkPipeline特點(diǎn)?并發(fā)高，Transform需要高并發(fā)滿足CPU消耗，同時Transform和Sinkchain到一起避免Shu?e開銷?一個Pipeline有多個表并且數(shù)量動態(tài)變化，無法對單表進(jìn)行并發(fā)?秒級實(shí)時，導(dǎo)入頻率高，從上游消費(fèi)到StarRocks數(shù)據(jù)可見保證秒級技術(shù)挑戰(zhàn)存算一體(Shared-nothing)存算分離(Shared-data)?sink.buffer-flush.max-bytes控制內(nèi)存使用?erval-ms控制導(dǎo)入延遲!!∞∞at-least-once/exactly-once即?按照merge_commit_interval_ms切分時間窗口，每個窗口內(nèi)的導(dǎo)入請求合并到一個事務(wù)?每個窗口部署一個分布式Plan，可以并行處理多個?時間窗口到達(dá)后Plan自動結(jié)束，提交當(dāng)前事務(wù)，下個時間窗?每個表通過merge_commit_?數(shù)據(jù)傳輸特點(diǎn)?數(shù)據(jù)傳輸特點(diǎn)?峰值并發(fā)導(dǎo)入請求高，約為表數(shù)*并發(fā)數(shù)，比如checkpoint觸發(fā)所有數(shù)據(jù)同時?ush，300表*300?高頻高并發(fā)下每個導(dǎo)入請求數(shù)據(jù)量少，導(dǎo)入請求可?為了避免請求之間相互阻塞，每個請求使用單獨(dú)TCP連接?TCP并發(fā)連接數(shù)過多帶來一些開銷和問題，比如TCPListenQueueOver?owTHANKYOU分析服務(wù)一體化（HybridServing/AnalyticsProcessing，HSAP）實(shí)時數(shù)倉實(shí)時數(shù)倉?百萬RPS實(shí)時寫入，寫入即可查?高性能實(shí)時整行更新和局部更新?高QPS、低延遲OLAP簡單查詢?復(fù)雜查詢：TPC-H30TB世界性能第一數(shù)據(jù)服務(wù)數(shù)據(jù)服務(wù)?百萬QPSKV點(diǎn)查，高可用?達(dá)摩院Proxima向量檢索?全SQL表達(dá)?行存、行列共存?OLAP分析和數(shù)據(jù)服務(wù)強(qiáng)隔離湖倉數(shù)據(jù)湖倉數(shù)據(jù)交互式分析?對離線數(shù)倉MaxCompute、數(shù)據(jù)湖進(jìn)行秒級交互式查詢?無需數(shù)據(jù)搬遷、元數(shù)據(jù)自動發(fā)現(xiàn)?百萬行每秒極速數(shù)據(jù)同步?內(nèi)外表聯(lián)邦分析hash(pk)pklsnXXpklsnXXXXpklsn12pklsnpklsndataXXXXXXpklsndata12pklsndata1pklsndata12pklsn12pklsn12pklsn12filebitmap?le1?le1pklsndata12pklsndata1pklsn12pklsn12pklsndata12pklsn12pklsndata12filebitmapfilebitmap?le1?le1pklsndata1pklsndata1lsnpklsnpkdatatype11INSERTINTOtable_nameVALUES(?,?,?),(?,?,?)ONCONFLICT(pk)DOUPDATE...aggressiveaggressive.enabled:truefrontendfrontendconngatewayconngateway?xed?xedfrontendjdbc:postgresql://HOST:PORT/DB?options=type=?xedsdkModesdkMode:jdbc_fixed?le1?le1INSERTINTOtable_nameVALUES(?,?,?),(?,?,?)ONCONFLICT(pk)DOUPDATE...WHEREexcluded.update_time>table_name.update_time;pklsnupdate_time12pklsndataupdate_time12INSERTINTOtable_nameVALUES(?,?,?),(?,?,?)ONCONFLICT(pk)DOUPDATE...clientserversqlsqlclientserversdkModesdkMode:jdbc_copyCOPYtable_nameFROMSTDINWITH(STREAM_MODEtrue,ON_CONFLICT'update');clientserverSourcerepartitionsinkSourcerepartitionsinkreshufflereshuffle-by-holo-distribution-key.enabled:truereshreshu?e-by-holo-distribution-key.enabled:truesdkMode:jdbc_fixedsdkMode:jdbc_fixedsdkMode:jdbc_copyarrowarrowarrowarrowarrow(serverless) 阿里云上客戶案例 阿里云上客戶案例阿里巴巴集團(tuán)案例輕松籌飛書深諾THANKYOUTheexplorationprocessofDinkyandFlinkCDCinreal-timefulldatabasesynchronization發(fā)生異常1111table1table1!!table2table1table2一table1table1SchemaChangeEventSchemaChangeEvent!!table2table1table2table2一一!一一一YamlYaml部署