腾讯实时流计算平台演进之路-杨华

杨华(vinoyang)腾讯高级工程师•Flink在腾讯实时计算概况简介•Oceanus平台简介•针对Flink的扩展与优化•Q&AFlink在腾讯的演进历程2017年上Flink框架预研，跟Storm进行对比2017年下Flink内部版本定制开发，Storm业务迁移，Standalone集群模式运行2018年上Flink产品化，打造一体化的实时流计算平台Oceanus,FlinkonYarn2018年下实时流计算平台规模化接入腾讯内部业务(覆盖所有BG)与外部客户并上线公有云2019年上完善场景化服务，上线在线ML，同时发力Flink批处理，打造SuperSQLFlink赋能腾讯实时计算峰值算⼒Flink在腾讯实时计算的规模集群总核数•Flink在腾讯实时计算概况简介•Oceanus平台简介•针对Flink的扩展与优化•Q&AOceanus平台整体技术架构Oceanus-应用列表Oceanus-画布详情Oceanus-指标统计Oceanus-在线机器学习Oceanus-在线机器学习•Flink在腾讯实时计算简介•Oceanus平台简介•针对Flink的扩展与优化•Q&AFlinkWebUI重构FlinkWebUI重构ZooKeeperdershipChanged3.ZooKeeperdershipChanged3.ReportTaskStateJobManager4.RecheckrunningstateTaskManagerTaskTaskTaskTaskManagerTaskTaskTaskJobManager3.ReportTaskStateJobManagerFailover优化TTaskManagerTaskTaskTaskStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierJobManagerChCheckpointinatorStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskJobManagerChCheckpointinatorStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierJobManagerChCheckpointinatorStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskStatefulTaskStatefulTaskTriggerCheckpointTaskManagerCheckpointBarrier StatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskJobManagerChCheckpointinatorStatefulTaskTaskManagerTaskManagerTriggerCheckpointCheckpointBarrierCheckpointinatorJobManagerTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierTriggerCheckpointTaskManagerStatefulTaskStatefulTasStatefulTaskTaskManagerTaskManagerTriggerCheckpointCheckpointBarrierCheckpointinatorJobManagerTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierTriggerCheckpointTaskManagerStatefulTaskStatefulTaskCheckpointBarrierStatefulTasStatefulTaskStatefulTasStatefulTaskStatefulTaskStatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpoint类重构:删除CheckpointTriggerResultStatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpoint类重构:删除CheckpointTriggerResult统一检查点异常类CheckpointDeclineReason->CheckpointFailureReasondingCheckpointabortXXXCheckpointBarrierJobManagerChCheckpointinatorStatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpointCheckpointBarrier类重构:删除StatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpointCheckpointBarrier类重构:删除CheckpointTriggerResult统一检查点异常类CheckpointDeclineReason->CheckpointFailureReasondingCheckpointabortXXXJobManagerChCheckpointinatorChCheckpointFailureManagerStatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointFailureManagerCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpointCheckpointBarrier类重构StatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointFailureManagerCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpointCheckpointBarrier类重构:删除CheckpointTriggerResult统一检查点异常类CheckpointDeclineReason->CheckpointFailureReasondingCheckpointabortXXXJobManagerChCheckpointinatorkpointFailureStatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointFailureManagerCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpointCheckpointBarrier类重构StatefulTaskStatefulTaskTaskManagerTriggerCheckpointCheckpointBarrierStatefulTaskTaskManagerStatefulTaskTriggerCheckpointCheckpointFailureManagerCheckpointBarrierTriggerCheckpointStatefulTaskTaskManagerStatefulTaskCheckpointBarrierTriggerCheckpointCheckpointBarrier类重构:删除CheckpointTriggerResult统一检查点异常类CheckpointDeclineReason->CheckpointFailureReasondingCheckpointabortXXXJobManagerChCheckpointinatorkpointFailurewindowwatermark大于windowwatermark大于hancedWindowttttteventstreamttttteventstream小于(丢弃)wwindowwatermark小于大于hancedWindow指标统计场景INSERTINTOt_minute_topic_cntCTtopicsum(cnt)ASsort_cnt,fixedTime(ENHANCED_START(pkgTime,INTERVAL'60'SECOND),'yyyyMMddHHmm')FROMtdsort_packcnt_flinkUPBYENHANCED(pkgTime,INTERVAL'60'SECOND),topic5minR(n+m)=5minR(n+m)=R(n+m-1)+delta5minR(n)=R(n-1)+deltaentWindowEEventeeeeeeeeeeee5minR(n+1)=R(n)+deltaSinkR(n)SinkR(n+1)24hSinkR(n+m)PurgeR(n+m)entWindow一天中游戏用户钻石消耗小时粒度的增长趋势SELECTuserId,SQL用法：SUM(unitsSQL用法：INCREMENT_TIME(true)FROMConsumesGROUPBYINCREMENT(consumeTime,INTERVAL'1'DAY,INTERVAL'1'hour),userIdTableAPI用法：valinput=env.addSource(newSourceFunction[Tuple2[String,Long]](){......}).assignTimestampsAndWatermarks(newAssignerWithPeriodicWatermarks[(String,Long)]{......}).toTable(tEnv,'a,'b,'cTableAPI用法：valwindowedTable=input.window(Incrementover1.dayevery1.houron'cas'w).groupBy('w,'a).select('b.sum,'a.count,incrementTime(true)) 4553 4553122 43355 455LocalKeyBy缓解数据倾斜4455Source->KeyBy->Count->Sink 223331Source->LocalKeyBy->Window->Count->KeyBy->Sum->SinkocalKeyByVSKeyByTask1activeKeyBy/WindowTaskTask1activeKeyBy/WindowTaskNAssignWatermark优化:下游算子检测IdleKafkapartition1KafkapartitionNassignwatermarkKeyByKeyBy/WindowTask1TaskNTask1Keyby/WindowTask1activeTask1Keyby/WindowTaskNTask1Keyby/WindowTask1activeTask1Keyby/WindowTaskNTaskNMapTaskNAssignWatermark优化:下游算子检测IdleKafkapartition1KafkapartitionNa