FFA2024分论坛-数据集成 合辑

FlinkCDC是基于数据库的日志CDC（ChangeDataCapture）技术，流式数据集成框架，配合Flink优秀的管道能力和丰富的上下游生态，FlinkCDC可以高效实现海量数据的实时集成。全量数据实时,一致性Paimon增量数据Paimon增量数据…Iceberg FlinkSQLAPIFlinkDataStreamAPIaggregatejoinJOIN链路组件多数据新鲜度数据一致性研发技术栈链路组件多数据新鲜度数据一致性研发技术栈…端到端作业不丢不重亚秒级延迟端到端作业不丢不重亚秒级延迟…Iceberg FlinkSQLAPIFlinkDataStreamAPIaggregatejoinJOIN全增量一体化2020/072021/082022/112023/102024/012024/09tabletable一Task2Task3Task1FlinkCDC3.x：端到端实时数据集成框架Analytics/BIDatabaseDataLakeDataWarehouse息StarRocksStarRocksPaimon$>flink-cdc$>flink-cdc.shmysql-to-starrocks.yaml mysql-to-starrocks.yamlDebeziumDebeziumInsertAfterCreateTableEventInsertAfterDataChangeEventInsertAfterCreateTableEventInsertAfterDataChangeEventAddColumnEventTruncateTableEvent…SchemaChangeEventAddColumnEventTruncateTableEvent…SchemaChangeEventStreamRecordStreamRecord1U13311U1331U133DataChangeEventSchemaChangeEventSchemaChangeEventBinaryData+SchemalessIDLEWAITINGIDLEWAITINGFINISHEDAPPLYINGFINISHEDAPPLYINGSchemaregistryappliesschemachangeSchemaChangeEventDataChangeEventSchemaChangeEventDataChangeEventFlushEventSchemaregistryacceptsschemachange,andrejectsfollowingrequestsFlushEventSchemaregistryacceptsschemachange,andrejectsfollowingrequestsSchemaoperatorregistersschemachangethenwaitforresponse(holdupstream).BlocksifSchemaRegistryisbusySinknotifiesflushcompleteSchemaregistryconfirmsschemaevolutioncompletes,readyfornextrequestSchemaoperatorbroadcastsFlushEvent,requestsregistryagaintowaitforflushcompleteSchemaoperatorreceivesSchemaChangeEventSchemaoperatorreleasesupstreamtherschemaoperatormustwaituntilotherschemachangesare SchemaOperatorDataSource SchemaOperatorDataSource PostPartitionerDataSinkPaimonPaimonStarRocks………l… 业务场景：•RAG(检索增强生成)00 0 0 支持AIModel（已有PR）支持Batchpipeline对接更多上下游，如：Iceberg，ClickHouse支持更多类型，包括schemachange类型和eventchange类型拥有独立的文档网站:/flink/flink-cdc-docs-stableTHANKYOU全行拥有273家分支机构网点覆盖长三角、珠三角环渤海湾等发达经济圈省内地市实现网点全覆盖杭州银行成立于1996年9月，是一家典型的城市商业银行，坚持服务区域经济、中小企业和城乡居民的市场定位，致力于为客户提供专业、便捷、亲和、全面的金融服务2016年10月27日，在上海证券交易所成功挂牌上市，截至2023年底，总资产18413.42亿元，2023年营业收入350.16亿元，利润收入162.82亿元。l在英国《银行家》杂志2023年公布的全球银行1000l2023年《财富》中国500强企业位列第380位l连续蝉联“十佳城商行”l最具竞争力中小银行l最佳普惠金融成效奖l浙江省融资畅通工程“突出贡献奖”l最佳科技金融服务城商行l支持浙江省经济社会发展优秀单位一等奖40l获评“杭州市模范集体”DBlink初探FlinkCDC20242023201720072024202320172007DataX增强FlinkCDC需要更快的同步数据，提高数….…..….…..理 高可用高可用yarnyarn….拿批流拿Case1:基于watermark自动生成Tag….拿批流拿….….↓↓Catalog/Database/TableCatalogBaseCatalog/Database/TableCatalogBase否数据开发方否是采集平台方否数据开发方否是采集平台方源系统方_」_」!!JavaClientJavaClient!!!THANKYOU 完全自主研发的数据库，才能做到核心替代的真正落地完全自主研发的数据库，才能做到核心替代的真正落地—原生多租户架构—自研一体化架构兼容经典模式，实现单机和分布式、TP和AP的融合—原生多租户架构—自研一体化架构兼容经典模式，实现单机和分布式、TP和AP的融合—集中式/分布式一体化架构——集中式/分布式一体化架构——基于Paxos+数据同步的灵活的容灾架构——基于Paxos+数据同步的灵活的容灾架构—★低成本—灵活的部署模式——灵活的部署模式—★多租户用纯列存表行列冗余表纯列存表createtablet2(c1createtablet2(c1varchar(1),c2varchar(1))withcolumngroup(eachcolumn);c1varchar(1),c2varchar(1))withcolumngroup(allcolumns,eachcolumn);语法说明语法说明l基于资源组的资源隔离方案l基于资源组的资源隔离方案lAP业务:多个AP业务的处理优先级不同基于资源组的资源隔离方案基于资源组的资源隔离方案…..…..…..…..…..…..••快速执行路径•优化写入放大问题•表锁阻塞写入insert/*+appendenable_parallel_dmlparallel(3)*/intot1select……;loaddata/*+direct(true,1024)parallel(16)*/infile……;Tablet’04xlarge(16c32G)图图…….★不仅仅是KV★不仅仅是KV★通用的使用场景★易用的分布式存储★易用的分布式存储基础类型HBase基础类型HBase融合查询并行执行融合查询并行执行多值索引全文索引多值索引全文索引多分区水平扩展高效存储读写分离多分区水平扩展高效存储读写分离•提供•提供SQL客户端和PythonSDK两种使⽤⽅式建表语句，向量作为一个单独的列createtablet1(c1vector(3),c2int,c3float,primarykey(c2));数据导入，使用标准SQL语法导入insertintot1values('[7576.42,467.23,2913.762]',3,54.6);使用DDL进行SQL语法，可以指定索引类型和距离算法CREATEINDEXvidx_c1_t1ont1(c1l2)usinghnsw;带有向量计算的SQL，优化器会自动使用向量索引OceanBase显著增强了服务的稳定性。OceanBase的分区表特性和旁路导入特性的分区表特性和旁路导入特性•对于分区表，按分区写入数据，减少分布式事务，提高写入效率•经过内部测试和社区用户使用验证，优化后的写入性能是普通SQL批量写入的3倍以上数据数据buffer根据分区攒批计算RowData分区并发写入单分区数据OceanBaseOceanBasedirectLoader.begin!bashbashbin/flink-cdc.shmysql-to-starrocks.yamlSchemaChangeEventOceanBasePipelineSink•管控功能齐全•适配数据源类型比较少数据源需要从头开发OceanBase•管控功能齐全•适配数据源类型比较少数据源需要从头开发OceanBase目前OceanBase正在基于自维护的Flink写连接器开发命令行工具，期望实现对FlinkCDC和FlinkJDBC连接器的数据源接入。后续该工具也将集成到OMS当中，供社区用户使用。OceanBase数据库基于列存引擎、并行执行引擎、向量化引擎，以及全新的优化器，具备担任轻型数仓的能力，数仓场景可以全部使用OceanBase数据库来简化ETL。OceanBase列存表OceanBase列存表/blog/12393391987 完善Flink命令行工具，实现 OceanBase多模API（HBase、Redis）全面适配 OceanBase向量数据库能力THANKYOUApacheAmoroPPMC&社区运营ApacheFlinkContrubitor&社区志愿者用配衡长低中高 ............化机制，为用户提供了解决三方悖论的方案。与传统的数据仓库不同,Lakehouse表格被广泛应用于分析师和管理员至关重要，Amoro通过在流计算中采用水印概念来解决这一挑战，从而评估表格新鲜度。•数据Schemachange题•....fragmentfragment/segminsert,eq-delete,pos-deleteinsert,pos-deletefragment,segmentsegmentinsert,eq-delete,pos-deleteinsert,pos-deletefragment,segmentsegmentinsert,eq-delete,pos-deleteinsertIceberg有主键表,既支持了ODS层,继续构建下游的表；尤其是还可以数据写入Kafka，给下游提供毫秒级延迟的数据，避免数据流经多个层级后延迟不断叠加0430THANKYOU ODSODSKafkaKafkaKa-aTBase(Ka-aTBase(PG)23445TopicKafkaSourceTopicKafkaSource地市1Topic多个分析数据库地市1Topic多个分析数据库Paimon地市2地市nTHANKYOU•增量计算(UPSERT+DELETE)典型场景：网络安全、电信运营商FlinkPipeline特点•并发高，Transform需要高并发满足CPU消耗，同时Transform和Sinkchain到一起避免Shuffle开销•一个Pipeline有多个表并且数量动态变化，无法对单表进行并发•秒级实时，导入频率高，从上游消费到StarRocks数据可见保证秒级技术挑战存算一体(Shared-nothing)存算分离(Shared-data)•sink.buffer-flush.max-bytes控制内存使用•erval-ms控制导入延迟!!∞∞at-least-once/exactly-once即•按照merge_commit_interval_ms切分时间窗口，每个窗口内的导入请求合并到一个事务•每个窗口部署一个分布式Plan，可以并行处理多个•时间窗口到达后Plan自动结束，提交当前事务，下个时间窗•每个表通过merge_commit_•数据传输特点•数据传输特点•峰值并发导入请求高，约为表数*并发数，比如checkpoint触发所有数据同时flush，300表*300•高频高并发下每个导入请求数据量少，导入请求可•为了避免请求之间相互阻塞，每个请求使用单独TCP连接•TCP并发连接数过多带来一些开销和问题，比如TCPListenQueueOverflowTHANKYOU分析服务一体化（HybridServing/AnalyticsProcessing，HSAP）实时数仓实时数仓•百万RPS实时写入，写入即可查•高性能实时整行更新和局部更新•高QPS、低延迟OLAP简单查询•复杂查询：TPC-H30TB世界性能第一数据服务数据服务•百万QPSKV点查，高可用•达摩院Proxima向量检索•全SQL表达•行存、行列共存•OLAP分析和数据服务强隔离湖仓数据湖仓数据交互式分析•对离线数仓MaxCompute、数据湖进行秒级交互式查询•无需数据搬迁、元数据自动发现•百万行每秒极速数据同步•内外表联邦分析hash(pk)pklsnXXpklsnXXXXpklsn12pklsnpklsndataXXXXXXpklsndata12pklsndata1pklsndata12pklsn12pklsn12pklsn12filebitmapfile1file1pklsndata12pklsndata1pklsn12pklsn12pklsndata12pklsn12pklsndata12filebitmapfilebitmapfile1file1pklsndata1pklsndata1lsnpklsnpkdatatype11INSERTINTOtable_nameVALUES(?,?,?),(?,?,?)ONCONFLICT(pk)DOUPDATE...aggressiveaggressive.enabled:truefrontendfrontendconngatewayconngatewayfixedfixedfrontendjdbc:postgresql://HOST:PORT/DB?options=type=fixedsdkModesdkMode:jdbc_fixedfile1file1INSERTINTOtable_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:truereshreshuffle-by-holo-distribution-key.enabled:truesdkMode:jdbc_fixedsdkMode:jdbc_fixedsdkMode:jdbc_copyarrowarrowarrowarrowarrow(serverless) 阿里云上客户案例 阿里云上客户案例阿里巴巴集团案例轻松筹飞书深诺THANKYOUTheexplorationprocessofDinkyandFlinkCDCinreal-timefulldatabasesynchronization发生异常1111table1table1!!table2table1table2一table1table1SchemaChangeEventSchemaChangeEvent!!table2table1table2table2一一!一一一YamlYaml部署