騰訊實(shí)時(shí)流計(jì)算平臺(tái)技術(shù)架構(gòu)

1、技術(shù)創(chuàng)新，變革未來騰訊實(shí)時(shí)流計(jì)算平臺(tái)技術(shù)架構(gòu)目錄Flink在騰訊實(shí)時(shí)計(jì)算概況簡(jiǎn)介Oceanus平臺(tái)簡(jiǎn)介針對(duì)Flink的擴(kuò)展與優(yōu)化Q&AFlink在騰訊的演進(jìn)歷程2017年上Flink框架預(yù)研，跟Storm進(jìn)行對(duì)比2017年下Flink內(nèi)部版本定制開發(fā)，Storm業(yè)務(wù)遷 移，Standalone集群模式運(yùn)行2018年上Flink產(chǎn)品化，打造一體化的實(shí)時(shí)流 計(jì)算平臺(tái)Oceanus,Flink on Yarn2018年下實(shí)時(shí)流計(jì)算平臺(tái)規(guī)?；尤腧v訊 內(nèi)部業(yè)務(wù)(覆蓋所有BG)與外部客 戶并上線公有云2019年上完善場(chǎng)景化服務(wù)，上線在線ML， 同時(shí)發(fā)力Flink批處理，打造 SuperSQLFlink

2、賦能騰訊實(shí)時(shí)計(jì)算Flink 在騰訊實(shí)時(shí)計(jì)算的規(guī)模集群總核數(shù)34萬2.1億/s峰值算力近20萬億日均消息量目前，騰訊內(nèi)部除廣告的在線訓(xùn)練業(yè)務(wù)外，原先運(yùn)行在Storm上的實(shí)時(shí)流計(jì)算業(yè)務(wù)都已逐步遷移至Flink平臺(tái)，廣告業(yè)務(wù)的遷移 計(jì)劃預(yù)期也在今年下半年完成。目錄Flink在騰訊實(shí)時(shí)計(jì)算概況簡(jiǎn)介Oceanus平臺(tái)簡(jiǎn)介針對(duì)Flink的擴(kuò)展與優(yōu)化Q&AOceanus平臺(tái)整體技術(shù)架構(gòu)Oceanus-應(yīng)用列表Oceanus-畫布詳情Oceanus-指標(biāo)統(tǒng)計(jì)Oceanus-在線機(jī)器學(xué)習(xí)Oceanus-在線機(jī)器學(xué)習(xí)目錄Flink在騰訊實(shí)時(shí)計(jì)算簡(jiǎn)介Oceanus平臺(tái)簡(jiǎn)介針對(duì)Flink的擴(kuò)展與優(yōu)化Q&AFlink

3、Web UI 重構(gòu)Flink Web UI 重構(gòu)JobManager Failover 優(yōu)化ZooKeeperJobManagerJobManagerTaskManagerTaskTaskTask1. Grant LeadershipTaskManagerTaskTaskTaskTaskTaskManagerTaskTask3. Report Task State4. Recheck running state2. Notify Leadership Changed3. Report Task State檢查點(diǎn)重構(gòu)-當(dāng)前機(jī)制StatefulTaskStatefulTaskTaskManager

4、StatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerCheckpoint CoordinatorJobManagerTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrier檢查點(diǎn)重構(gòu)-當(dāng)前機(jī)制StatefulTaskStatefulTaskTaskManagerStatefulTaskState

5、fulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerCheckpoint CoordinatorJobManagerTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrier檢查點(diǎn)重構(gòu)-當(dāng)前機(jī)制StatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManage

6、rStatefulTaskStatefulTaskTaskManagerCheckpoint CoordinatorJobManagerTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrierFail Job檢查點(diǎn)重構(gòu)-當(dāng)前機(jī)制StatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerStateful

7、TaskStatefulTaskTaskManagerCheckpoint CoordinatorJobManagerTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrierFail Job引入CheckpointFailureManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerS

8、tatefulTaskStatefulTaskTaskManagerJobManagerCheckpoint CoordinatorTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrier引入CheckpointFailureManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerSt

9、atefulTaskStatefulTaskTaskManagerJobManagerCheckpoint CoordinatorTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrier類重構(gòu):刪除CheckpointTriggerResult 統(tǒng)一檢查點(diǎn)異常類CheckpointDeclineReason - CheckpointFailureReason PendingChec

10、kpoint#abortXXX引入CheckpointFailureManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrierJobManagerCheckpoi

11、nt CoordinatorCheckpoint FailureManagerFailure counter類重構(gòu):刪除CheckpointTriggerResult 統(tǒng)一檢查點(diǎn)異常類CheckpointDeclineReason - CheckpointFailureReason PendingCheckpoint#abortXXX引入CheckpointFailureManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerJ

12、obManagerCheckpoint CoordinatorTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckpointBarrierCheckpointBarrierCheckpointBarrierReport Checkpoint FailureCheckpoint FailureManagerFailure counter類重構(gòu):刪除CheckpointTriggerResult 統(tǒng)一檢查點(diǎn)異常類CheckpointDeclineReason - Chec

13、kpointFailureReason PendingCheckpoint#abortXXX引入CheckpointFailureManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerStatefulTaskStatefulTaskTaskManagerJobManagerCheckpoint CoordinatorTriggerCheckpointTriggerCheckpointTriggerCheckpointTriggerCheckpointCheckpointBarrierCheckp

14、ointBarrierCheckpointBarrierCheckpointBarrierReport Checkpoint FailureCheckpoint FailureManagerFailure counter類重構(gòu):刪除CheckpointTriggerResult 統(tǒng)一檢查點(diǎn)異常類CheckpointDeclineReason - CheckpointFailureReason PendingCheckpoint#abortXXXEnhanced Windowevent streamwindowcurrent watermark大于小于(丟棄)ttttevent streamwi

15、ndowcurrent watermark大于小于ttttEnhanced WindowINSERT INTO t_minute_topic_cnt SELECT topic,sum (cnt) AS sort_cnt,fixedTime (ENHANCED_START(pkgTime, INTERVAL 60 SECOND), yyyyMMddHHmm) FROM tdsort_packcnt_flinkGROUP BYENHANCED (pkgTime, INTERVAL 60 SECOND), topic指標(biāo)統(tǒng)計(jì)場(chǎng)景Increment WindowEventeeee.eeeeeeee5m

16、in.5min5minR(n) = R(n-1) + deltaSink R(n)R(n+1) = R(n) + deltaSink R(n+1)R(n+m) = R(n+m-1) +deltaSink R(n+m)24hPurge R(n+m)Increment WindowSELECT userId,SUM(units), INCREMENT_TIME(true)FROM ConsumesGROUP BY INCREMENT(consumeTime, INTERVAL 1 DAY, INTERVAL 1 hour), userId一天中游戲用戶鉆石消耗小時(shí)粒度的增長(zhǎng)趨勢(shì)SQL用法：val

17、input = env.addSource(new SourceFunctionTuple2String,Long().).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks(String, Long) .).toTable(tEnv, a, b, c.rowtime)Table API用法：val windowedTable = input.window(Increment over 1.day every 1.hour on c as w).groupBy(w, a).select(b.sum, a.count,

18、 incrementTime(true)LocalKeyBy緩解數(shù)據(jù)傾斜455455Source - KeyBy - Count- Sink322133323Source - LocalKeyBy - Window - Count - KeyBy - Sum - Sink132455455LocalKeyBy VS KeyByAssign Watermark優(yōu)化:下游算子檢測(cè)IdleSource Task 1Source Task NKafka partition1Kafka partitionNKeyBy/WindowTask 1KeyBy/WindowTask N.activeidleas

19、sign watermarkSource Task 1Source Task NKafka partition1Kafka partitionNMap Task 1Map Task N.assign watermarkFilter Task 1Filter Task N.activeidleKeyby/ Window Task 1Keyby/ Window Task NNo data, No watermarkAssign Watermark優(yōu)化:下游算子檢測(cè)Idle日志重構(gòu)：日志展示與日志分離UserClassLoader重構(gòu)與日志分離User Classloader1Task Manage

20、rUser Classloader2Task for job2Task for job1Logger LOG = LoggerFactory.getLogger();Logger LOG = LoggerFactory.getLogger();-Dlog4j.configuration=-Dlog4j.file=Parent Classloaderorg.slf4j.Logger org.apache.log4jUserClassLoader重構(gòu)與日志分離Parent Classloaderorg.slf4j.Loggerorg.apache.log4jUser Classloader1 org.slf4j.Logger org.apache.log4jTask Manager