1、FlinkBatch SQL Improvements on Lakehouse刘大龙/阿里云研发工程师Streaming lakehouse meetupFlink Batch on Paimon 挑战0101CONTENT目录 0202Flink Batch 核心优化0303后续规划01Flink Batch on Paimon 挑战Streaming Warehouse:Flink+PaimonLogsRDBMSFlink Table StoreFlink Table StoreFlink SQLStreaming&BatchFlink SQLStreaming&BatchbinlogD

2、ata ServingSystemsFlink SQLQueriesFlink Table StoreODSDWDDWSADSFlink SQLStreaming&BatchPaimonPaimonPaimonFlink CDC架构简洁语义统一数据一致成本低廉透明开放Flink Batch 挑战Schema 变更行级更新与删除Snapshot 管理时间旅行查询高效 ETL&Ad-hoc02Flink Batch 核心优化Year Recap of Apache Flink BatchFlink 1.16Flink 1.17Flink 1.18SQL GatewayAutomatic Colle

3、ction of StatisticsDynamic Partition PruningJoin HintAdaptive Hash JoinSpeculative ExecutionUpdate&DeleteDPP Strategy OptimizeBushy Join ReorderAdaptive Local HashAggAdaptive Batch SchedulerLakehouse APIsFlink JDBC DriverRuntime FilterOperator Fusion Codegen2022.102023.032023.09Part1:Lakehouse API E

4、nhanceALTER TABLE(FLINK-21634,FLINK-27237)CREATE/REPLACE TABLE AS SELECT(FLIP-218,FLIP-305,FLIP-303)Data Management APICALL Procedure(FLIP-311)Time Travel(FLIP-308)UPDATE/DELETE(FLIP-282)TRUNCATE TABLE(FLIP-302)Data Management APIPart2:Join 优化Statistics EnhanceAnalyze Table(FLIP-240)手动触发，持久化到 Catalo

5、g统计信息丰富rowCountnullCount,ndvmin,maxavgLen,maxLenSupportReportStatistics(FLIP-231)自动收集，不持久化，更实时Flink CSV&Parquet&ORC Format 已支持Paimon 已支持Planner 优先从 Catalog 中获取统计信息，没有则通过 SupportReportStatistics 方式实时获取45Join HintBroadcast Hash JoinBroadcast small table,build hash tableOnly support equi-join4545Shuffl

6、e Hash JoinSort Merge JoinNested Loop JoinBroadcast small table,spill to disk if too largeSupport both equi-join and no-equi-joinShuffle both side by join key and sortOnly support equi-joinShuffle both side by join key,build hash table use small tableOnly support equi-joinJoin Hint没有统计信息，Planner 给出的

7、 Join 算子不合理，慢慢慢！别慌，我们有 Join HintJoin HintBROADCASTSHUFFLE_HASHSHUFFLE_MERGENEST_LOOPFlink provide following join hint(FLIP-229):Join ReorderWhat is Join Reorder?以 TPC-DS q18 为例Join Reorder:Left-Deep Treecustomerics_demographcustomerics_demographdate_dimcatalog_salescustomercustomer_addressitemJoinJo

8、inJoinJoinJoinJoin1.首先第1层计算出每张表的 rowCount2.第2层从第1层中选出 rowCount 最小的表，Join 其它 N-1 张表，计算出每两个 Join 的输出 rowCount3.第K层从第K-1层中选出 rowCount 最小的 Join，Join 其它N-K+1张表，计算出K张表Join的输出 rowCount4.当K=N时，算法结束问题：贪心算法，容易陷入局部最优解Join 只能顺序执行，无法并行影响 DPPJoin Reorder:Bushy Treecustomer_addressdate_dimcustomerics_demographcata

9、log_salesJoincustomerics_demographitemJoinJoincustomerJoinJoinJoin优势：基于动态规划算法，整棵树更加平衡，全局最优更多pattern 可以 match 上 DPP根据时间和空间复杂度自适应切换两种算法Q18 性能提升一倍FLINK-30376Partition Tableslod_date=2000slod_date=1slod_date=2CREATE TABLE sales(sold_dateINT,item_idINT,sales_priceDOUBLE,list_priceDOUBLE,quantityINT,disco

10、unt_amtDOUBLE,wholesale_cost DOUBLE,ext_taxDOUBLE)PARTITIONED BY(sold_date)WITH(CONNECTOR=ORC);如何减少无效分区数据读取？Dynamic Partition PruningSELECT*FROM sales WHERE sold_date=2/分区字段上有过滤条件12ScanFiltersold_date=212Scansold_date=2优化阶段已经确定有效分区无分区裁剪静态分区裁剪Dynamic Partition PruningSELECT*FROM sales JOIN date_dimWH

11、ERE sold_date=date_sk AND year=2000/分区字段上无过滤条件Joinsold_date=date_skFilteryear=2000Scandate_dimCREATE TABLE date_dim(date_idINT,date_skINT,year INT,month INT,date VARCHAR)WITH(CONNECTOR=ORC);只有 year=2000 的 date_sk 可以关联出结果12Scansales过滤掉无效的分区FLIP-24812ScansalesSELECT*FROM sales JOIN date_dim WHERE item

12、_id=date_skJoinitem_id=date_skExchangeScandate_dimCREATE TABLE date_dim(date_idINT,date_skINT,year INT,month INT,date VARCHAR)WITH(CONNECTOR=ORC);Runtime FilterExchangeJoin key不包含分区字段，能不能减少Shuffle数据？12ScansalesJoinitem_id=date_skExchangeScandate_dimRuntime FilterRuntimeFilterdate_sk in set(item_id)E

13、xchangeFilterBuilderset(item_id)根据date_dim表item_id集合，提前过滤sales表的大量数据，提升Join算子的性能FLIP-324Part3:Rutime 优化Operator Fusion CodegenTPC-DS Q99 profiling（多路 Join）虚函数调用中间数据物化到内存Source Iterator开销collector开销无效计算Operator Fusion Codegen我们需要优化什么？尽量避免内存访问，数据驻留在寄存器消除虚函数调用编译执行，面向 query 生成最优的代码For 循环处理延迟计算 Thomas Ne

14、umanns paper:Efficiently Compiling Efficient Query Plans for Modern Hardware Collapse entire query into a single operator Generate one function for the entire query 批作业支持 OFCG(FLIP-315)Operator Fusion CodegenScanFilterProjectAggregatelong count=0;for(item in sales)if(price 100)count+=1;设置作业并行度？海量批处理

15、作业作业数据量每日变化中间节点数据量难以判断SQL 作业并行度全局统一费时费力调优困难资源浪费额外开销Adaptive Batch SchedulerAdaptive Batch SchedulerA1A2Vertex AIntermediate ResultVertex BB1B4（1）（2）（3）1x2x3x4x expected data per task改进效果（FLIP-187）更易用，无需为每个作业单独调整并行度更高的资源利用率自动负载均衡批作业默认批作业默认SchedulerScheduler（FLIPFLIP-283283）整体性能追齐主流批计算引擎TPC-DS Benchma

16、rk Evolution88736993527749430040005000600070008000900010000总时间（单位：s）TPCTPC-DSDS 10T10TFlink-1.16分区表Flink-1.16Flink-1.17Flink-1.18(TODO)Part4:稳定性优化Adaptive Hash Join问题：对于 Hybrid HashJoin，数据倾斜非常严重时，递归会无限循环下去，导致 Join 无法结束Hybrid HashJoin 算法流程Adaptive Hash Join对倾斜分区进行优化，当递归次数大于3时，转为 Sort Merge

17、 Join，既保证了性能，又提升了稳定性FLINK-26929Adaptive HashJoin 算法流程Node 3热点机器(I/O 繁忙，CPU 高负载)无法避免，其上执行的数据处理任务可能异常缓慢批作业产出时间难以得到保障预测执行预测执行为热点机器上的慢任务创建新的执行实例(attempts)，并部署在正常的机器节点上这些预测执行实例和对应的原始实例具有相同的输入和产出最先完成的实例会被承认，其他相应实例会被取消A1A2A3B1B2Node 1Node 2Speculative ExecutionSpeculative ExecutionNode 3A1A2A3B1B2Node 1Nod

18、e 2A3Node 4Speculative SchedulerSlowTaskDetectorwatch(1)report slow tasksBlocklistHandler(2)block slow nodes(5)cancel(4)finish(3)allocate slot and deploySource 支持预测执行（FLIP-245）保证同一个 Source 并发的不同执行实例读取相同的数据SourceFunction、SourceFormat、FLIP-27 Source 都已支持Sink 支持预测执行（FLIP-281）SinkFunction、SinkFormat、Sin

19、kV2 都已支持FLIP-168Part5:SQL 服务化SQL CLISQL Gateway 支持 HTTPS REST 协议(FLIP-91)SQL Gateway 支持 HiveServer2 协议(FLIP-223)Flink SQL CLI 支持连接到 SQL Gateway(FLIP-275)提供 Flink JDBC Driver，供应用程序连接到 SQL Gateway(FLIP-293)SQL Gateway03后续规划后续规划用户与生态引擎持续优化 现有功能持续完善，如 Runtime Filter、OFCG Format 读写性能优化 Adaptive Query Execution.Paimon&Hudi&Iceberg 等湖存储生态对接，数据湖分析场景发力 更多 Focus 在解决用户问题，打磨引擎（字节、快手、Shopee、小红书等）.ThanksStreaming lakehouse meetup