1、高赟（云骞）面向流批一体的面向流批一体的FlinkFlink RuntimeRuntime新进展新进展The New Features of the Runtime for Unified Batch and Stream Processing流批一体流批一体性能优化性能优化RemoteRemoteShuffleShuffle总结与展望总结与展望#1#2#3#4#5语义完善语义完善与增强与增强#1#1流批一体流批一体什么是流批一体什么是流批一体What is batch/stream unificationWhat is batch/stream unificationUnboundedBou

2、ndedDataStream APITable/SQL APIBoundedUnbounded为无限数据和有限数据的处理提供统一的APIUnified API for processing unbounded and bounded data为什么需要流批一体为什么需要流批一体Why batch/stream unificationWhy batch/stream unification 减少流处理与批处理的学习与维护成本 避免流批融合作业数据的不一致性-Backfill流批一体流批一体APIAPIUnified API for Batch/Stream ProcessingUnified A

3、PI for Batch/Stream ProcessingTable/SQL APIDataStream APIOperatorsEvent-timer/WatermarkProcessing TimerStateStandard SQL/Table如何实现流批一体如何实现流批一体APIAPIHow to implement batch/stream unificationHow to implement batch/stream unificationState流执行模式批执行模式基于State的增量计算逻辑上可以支持任意操作-保留所有历史数据+Retraction一些场景下存在一定性能问

4、题-特定算子缺少有效优化-增量计算导致State随机I/O-定时Checkpoint算子逐级运行依赖于数据有限性算子可以进行特定的优化-数据预先Sort，按Key逐个处理当给定相同的输入数据时，保证两种执行模式的一致性。FlinkFlink流批一体流批一体架构架构Unified Batch/Stream Processing in Unified Batch/Stream Processing in FlinkFlink统一PipelinedRegion调度统一ShuffleService插件接口统一ConnectorStream ModeBatch ModeDataStream APITab

5、le/SQL APIRuntimeExecution ModeAPI#2 2语义增强与完善语义增强与完善部分部分TaskTask结束后结束后CheckpointCheckpointSupportSupport CheckpointsCheckpoints afterafter SomeSome TasksTasks FinishedFinished如果所有Source都是有限的，那么作业最终会结束如果有些Source是无限的，用户可能通过stop-with-savepoint-drain终止作业-stop-with-savepoint 不属于终止1 FLIP-147:Support Check

6、points after Tasks Finished,https:/cwiki.apache.org/confluence/display/FLINK/FLIP-147%3A+Support+Checkpoints+After+Tasks+Finished部分部分TaskTask结束后结束后CheckpointCheckpointSupportSupport CheckpointsCheckpoints afterafter SomeSome TasksTasks FinishedFinished有限数据Source和无限数据Source的混合作业无法支持CheckpointStream模式

7、依赖两阶段提交来提交数据-数据首先写入临时文件或者事务中-Checkpoint成功后重命名文件或提交事务Stream模式下作业结束时最后一部分数据无法提交无法保证两种模式下作业的结果一致性部分部分TaskTask结束后结束后CheckpointCheckpointSupportSupport CheckpointsCheckpoints afterafter SomeSome TasksTasks FinishedFinished支持部分Task结束后进行Checkpoint作业结束前Task等待一个Checkpoint完成后退出-正常结束：等待下一个Checkpoint完成-Drain:等待

8、当前Savepoint完成后退出部分部分TaskTask结束后结束后CheckpointCheckpointSupportSupport CheckpointsCheckpoints afterafter SomeSome TasksTasks FinishedFinishedFully Finished识别新的Source结点发起Checkpoint记录全部完成状态重启后全部完成的节点直接结束-部分结束：Rescale部分部分TaskTask结束后结束后CheckpointCheckpointSupportSupport CheckpointsCheckpoints afterafter S

9、omeSome TasksTasks FinishedFinished结束流程1.Max Watermark2.On EndOfPartitionEvent2.1 endOfInput(i)2.2 close()2.3 dispose()Max watermarkEndOfPartitionEventBarrier for savepoint在这里等Checkpoint？Checkpoint CompleteDrain流程1.触发Savepoint2.Savepoint成功后Source结束3.On EndOfPartitionEvent3.1 endOfInput(i)3.2 close()

10、3.3 dispose()这里已经进行了一个Checkpoint？部分部分TaskTask结束后结束后CheckpointCheckpointSupportSupport CheckpointsCheckpoints afterafter SomeSome TasksTasks FinishedFinishedEndOfDataEvent统一结束流程1.Max Watermark2.On EndOfDataEvent2.1 endOfInput(i)2.2 close()finish()2.3 如果是Source&Drain，Trigger Savepoint2.4 等待下一个Checkpoi

11、nt/Savepoint结束3.On EndOfPartitionEvent3.1 dispose()close()Max watermarkBarrierEndOfPartitionEventEndOfPartitionEvent之后不能再发送CheckpointBarrier流批混合流批混合 SourceSourceHybrid SourceHybrid Source读取批数据后切换到无限数据适用于处理逻辑一致时的批流互转-实时数据已落盘，Backfill#3 3性能优化性能优化调度部署性能优化调度部署性能优化Optimization of SchedulingOptimization o

12、f SchedulingJobGraphExecutionGraphPointWiseAll to Alln2 条边 ExecutionGraph边的数量是n2 作业初始调度/Failover时间过长，内存占用过大 Batch平均作业规模更大，加重调度与部署性能问题1 https:/issues.apache.org/jira/browse/FLINK-21110调度性能优化调度性能优化Optimization of SchedulingOptimization of SchedulingConsumerPartitionGroupConsumerVertexGroupConsumerPart

13、itionGroupConsumerVertexGroupPointWiseAll to All 利用对称性，引入ConsumerGroup结构代替ExecutionEdge 调整调度算法，复用ConsumerGroup避免n2遍历调度性能优化调度性能优化Optimization of SchedulingOptimization of SchedulingPipelineBlockinga1c1b1d1a2c2d2b2 Flink流批统一Dag-Pipeline Shuffle:上下游同时启动，通过网络在线传输-Blocking Shuffle:上下游依次启动，通过文件离线传输 调度时需要按

14、Blocking边切分为多个Region 存在环形依赖问题-(a1,b1,c1,d1)(a2,b2,c2,d2)优化前：Tarjan强连通分量，O(n2)优化后：-在JobGraph层识别Pipeline Region-如果有all-to-all边，一定有环形依赖，直接合并 O(n)-如果没有all-to-all边，Tarjan强连通分量，O(n)部署性能优化部署性能优化Optimization of Scheduling and DeploymentOptimization of Scheduling and Deploymenta1:172.2.3.1a2:172.2.3.3a1:172.

15、2.3.5 Shuffle Descriptor Flink部署下游任务时携带Shuffle Descriptors O(n2)-计算ShuffleDescriptor可能卡死主线程，HB Timeout-内存占用过大 利用对称性，缓存ShuffleDescriptor 如果ShuffleDescriptor过大，使用BlobServer传输调度与部署性能优化效果调度与部署性能优化效果Benefits of Optimization the Scheduling and DeploymentBenefits of Optimization the Scheduling and Deployme

16、nt1.121.14作业初始化11,431 ms627 ms任务部署63,188 ms17,183 msFailover重启节点37,195 ms170 ms内存占用30GB2GB10000*10000,All to All单点重启单点重启SingleSingle-task Failover Strategytask Failover Strategy 避免单个Task失败时重启其它Task 提供不同的一致性语义-Best Effort-At-least-once-Exactly-once 进行中Buffer DebloatingBuffer DebloatingBuffer Debloati

17、ngBuffer Debloating 根据处理能力动态调整Buffer大小 减少Checkpoint时反压导致的长对齐时间1 FLIP-183 Dynamic Buffer Size Adjustment,https:/cwiki.apache.org/confluence/display/FLINK/FLIP-183%3A+Dynamic+buffer+size+adjustment#4 4Remote ShuffleRemote Shuffle为什么需要为什么需要RemoteRemote ShuffleShuffleWhyWhy RemoteRemote ShuffleShuffleTa

18、skManagerTaskManagerNettyNettyServerServerTaskTask需要使用本地磁盘，无法支持云原生存储计算分离的架构计算资源与存储资源耦合，无法单独伸缩文件服务的稳定性受Task与TaskManager稳定性的影响插件化插件化ShuffleShuffle APIAPIThe Pluggable Shuffle APIThe Pluggable Shuffle APIPartitionTrackerSchedulerShuffleMaster请求/释放Shuffle资源TaskTaskResultPartitionInputGateJMTMTMRemoteRem

19、ote ShuffleShuffle整体架构整体架构The Architecture of the RemoteThe Architecture of the Remote ShuffleShuffleSchedulerShuffleMaster请求/释放Shuffle资源TaskResultPartitionJMTaskInputGateAssignmentTrackingWorkerManagementDataManagementDataManagement请求/释放Shuffle资源请求/释放Shuffle资源Shuffle ManagerShuffle WorkerShuffle Wo

20、rkerRemoteRemote ShuffleShuffle数据读写数据读写Read/Write of the RemoteRead/Write of the Remote ShuffleShuffleTaskReadWriteTaskReadWriteTaskReadWriteRead ServiceWrite ServiceIO调度 压缩 Managed DirectMemory Zero Copy Credit-based 协议 TCP连接复用 IO调度优化，HDD IO速度 140MB/s 数据按下游合并开发中-MapPartition：单个上游给所有下游的数据-ReducePart

21、ition：所有上游给单个下游的数据RemoteRemote ShuffleShuffle使用使用TheThe DeploymentDeployment ofof thethe RemoteRemote ShuffleShuffle 同时支持Standalone/Yarn/K8s 多种方式 支持多版本协议兼容 提供Metric等运维工具-更多运维工具开发中项目地址：https:/ 5总结与展望总结与展望 Flink已具备可上线的流批一体处理能力 未来将进一步提升Flink流批一体处理能力-支持更多的流批融合的场景-Remote Shuffle的进一步完善与优化-流批处理模式下性能与稳定性的进一步优化 OLAP总结与展望Summary and Future Work20212021-1212-0505THANKS