Flink流批一體的技術(shù)架構(gòu)介紹

1、技術(shù)創(chuàng)新，變革未來(lái)Flink 流批一體的技術(shù)架構(gòu)介紹目錄需求和挑戰(zhàn)Flink 架構(gòu)簡(jiǎn)介流批一體的入口 SQL流批一體的大規(guī)模實(shí)踐 在線機(jī)器學(xué)習(xí)平臺(tái)總結(jié)和展望主流架構(gòu)對(duì)比LambdaKappa執(zhí)行引擎同時(shí)具備多種能力低延遲的流計(jì)算高吞吐、高穩(wěn)定性的批處理用戶角度：編程接口統(tǒng)一。一份代碼，一樣的結(jié)果開(kāi)發(fā)人員角度：架構(gòu)統(tǒng)一，代碼復(fù)用流批一體系統(tǒng)的需求和挑戰(zhàn)Flink 架構(gòu)簡(jiǎn)介StorageHDFS, HBase, KafkaLocalSingle JVMClusterStandalone, YARN, K8S, DeployDistributed Streaming DataflowRuntimeD

2、ataStream APIAPIDataSet APIWord Countvallines:DataStreamString= env.readFromQueue(address)valwords:DataStreamWord =lines.flatmap(line) = split(line)valcounts:DataStreamInt =words.keyBy(“word”).sum(“frequency”)counts.addSink(new RollingSink(path)計(jì)算模型的核心抽象表達(dá)作業(yè)邏輯的DAG主要由點(diǎn)和邊構(gòu)成點(diǎn)：算子（operator），包含主要的計(jì)算邏輯邊：數(shù)據(jù)

3、流通管道，可以運(yùn)行在多種介質(zhì)上（網(wǎng)絡(luò)、文件、內(nèi)存）Streaming Dataflowsource 1/2source 2/2flatmap 1/2flatmap 2/2aggregate 1/2aggregate 2/2sink 1/1Word Count：批處理是流計(jì)算的特例opUnbounded StreamUnbounded StreamopopopUnbounded StreamUnbounded StreamPipelineStream JobopopopopPipeline / PersistBounded StreamBounded StreamBounded StreamBo

4、unded StreamBatch Job流批統(tǒng)一的入口 - SQL-|USER_SCORES|-| User| Score | Time|-|Julie|7|12:01|Frank|3|12:03|Julie|1|12:03|Frank|2|12:06|Julie|4|12:07|-USER_SCORES is asource table/stream.Batch Mode:12:07 SELECTName, SUM(Score), MAX(Time)FROM USER_SCORES GROUP BY Name;-| Name| Score | Time|-| Julie | Frank |

5、12 | 12:07 |5 | 12:06 |-Stream Mode:12:01 SELECT Name, SUM(Score), MAX(Time)FROM USER_SCORES GROUP BY Name;12 | 12:07 | |5 | 12:06 | |8 | 12:03 | | | Julie |3 | 12:03 | | | Frank |-|-inf, 12:01)|12:01, 12:04)|12:04, now)| - | - | - | | Name| Score | Time| | | Name| Score | Time| | | Name| Score | Ti

6、me| | - | - | - | | | | | Julie | | | Frank | - | - | - |-流有 Early fire最終結(jié)果一致流批統(tǒng)一的架構(gòu)LocalSingle JVMCloudGCE, EC2ClusterStandalone, YARNRuntimeDistributed Streaming DataflowDataStream APIStream ProcessingDataSet APIBatch ProcessingTable API & SQLRelationalRuntimeDataStream APIStream ProcessingDataSet

7、 APIBatch ProcessingTransformationStreamGraphOperator TreeBatch PlanOptimized PlanJob GraphStream Task & OperatorBatch Task & Driver語(yǔ)義難以和 SQL 保持一致開(kāi)發(fā)效率低，添加功能鏈路長(zhǎng)執(zhí)行模式不同，代碼難以復(fù)用新架構(gòu)LocalSingle JVMCloudGCE, EC2ClusterStandalone, YARNRuntimeDistributed Streaming DataflowDataStream APIStream ProcessingDataSe

8、t APIBatch ProcessingTable API & SQLRelationalRuntimeDAG API & Stream OperatorsQuery ProcessorQuery Optimizer & Query ExecutorTable API & SQLRelationalLocalSingle JVMCloudGCE, EC2ClusterStandalone, YARN新架構(gòu)主要修改點(diǎn)RuntimeDAG API & Stream OperatorsQuery ProcessorQuery Optimizer & Query ExecutorTable API

9、& SQLRelationalLocalSingle JVMCloudGCE, EC2ClusterStandalone, YARNTable API 和 SQL 變成一級(jí) API引入 Query Processor 模塊統(tǒng)一流和批的處理使用相同的 DAG 和 Stream Operator 來(lái)描述流批作業(yè)Runtime 統(tǒng)一到流上 push based 實(shí)現(xiàn)未來(lái)可以考慮和 DataStream 共享算子Query Processor 簡(jiǎn)介大部分共用部分共用完全一樣SQL& Table APILogical PlanOptimizerPhysical PlanExecution DAG執(zhí)行引擎

10、同時(shí)具備多種能力低延遲的流計(jì)算高吞吐、高穩(wěn)定性的批處理用戶角度：編程接口統(tǒng)一。一份代碼，一樣的結(jié)果開(kāi)發(fā)人員角度：架構(gòu)統(tǒng)一，代碼復(fù)用回顧需求大規(guī)模實(shí)踐 在線機(jī)器學(xué)習(xí)平臺(tái)Event：用戶行為（商品 曝光、點(diǎn)擊、購(gòu)買等）Entity：準(zhǔn)靜態(tài)特征（商品7天的點(diǎn)擊數(shù)量等）Sample：樣本Event 和 Entity 的存儲(chǔ)選擇時(shí)效性：ETL 作業(yè) + 實(shí)時(shí)訓(xùn)練作業(yè)樣本一致性：規(guī)避 Early fire 導(dǎo)致的錯(cuò)誤樣本數(shù)據(jù)可回溯：任意環(huán)節(jié)的糾錯(cuò)能力批和流樣本的統(tǒng)一生成模型的 Validation 機(jī)制問(wèn)題和挑戰(zhàn)需要同時(shí)具備低延遲的流式訂閱和高吞吐的歷史數(shù)據(jù)回溯功能ETL 作業(yè)延遲和數(shù)據(jù)重復(fù)的權(quán)衡Exac

11、tly once 由于 Checkpoint barrier 對(duì)齊的問(wèn)題導(dǎo)致延遲波動(dòng)At least once 不保證數(shù)據(jù)不重復(fù)輸出解決方案：消息隊(duì)列 + 類 HBase 的 KV 系統(tǒng)消息隊(duì)列提供低延時(shí)的流式訂閱ETL 作業(yè)使用 At least once，利用 KV 系統(tǒng)的 Update 能力進(jìn)行去重并提供歷史數(shù)據(jù) Scan 功能平臺(tái)對(duì)數(shù)據(jù)源進(jìn)行包裝，在不同場(chǎng)景下切換存儲(chǔ)的選擇在 CVR（Conversion Rate，轉(zhuǎn)化率）模型中，我們會(huì)根據(jù)點(diǎn)擊是否成交而生 成相應(yīng)的樣本：假如用戶點(diǎn)擊后進(jìn)行了購(gòu)買，則是一個(gè)正樣本假如用戶點(diǎn)擊后沒(méi)有購(gòu)買，則是一個(gè)負(fù)樣本用戶從點(diǎn)擊到成交的時(shí)間不確定，從秒級(jí)

12、到小時(shí)級(jí)不等解決方案：利用 SQL 的 Retraction 機(jī)制在延遲容忍程度內(nèi)，先根據(jù)當(dāng)前數(shù)據(jù)情況輸出結(jié)果當(dāng)結(jié)果需要有變化時(shí)，先發(fā)送一條之前錯(cuò)誤的結(jié)果，標(biāo)記為 Retraction，然后再發(fā)送一條最新的正確結(jié)果用戶在算法邏輯中對(duì) Retraction 消息進(jìn)行處理，對(duì)錯(cuò)誤樣本進(jìn)行修正實(shí)時(shí)訓(xùn)練的時(shí)效性和一致性雖然有了一定的錯(cuò)誤修正機(jī)制，但還是避免不了產(chǎn)生一些負(fù)面影響。需要定 期進(jìn)行樣本的批量生成有些模型還不需要很高的時(shí)效性，更關(guān)注樣本準(zhǔn)確性解決方案：直接復(fù)用實(shí)時(shí)的樣本生成邏輯，一樣的 SQL，一樣的 UDF平臺(tái)將 Source 自動(dòng)替換成 KV 系統(tǒng)的歷史數(shù)據(jù)進(jìn)行 ScanSQL 作業(yè)自動(dòng)切

13、換成批處理模式執(zhí)行批處理作業(yè)使用混部資源運(yùn)行批流一體的樣本生成機(jī)制優(yōu)化資源使用避免 N*M 級(jí)別（上游并發(fā)為N，下游并發(fā)為M）的內(nèi)存占用消耗Metrics reporter 的內(nèi)存占用優(yōu)化穩(wěn)定性提升JobManager Failover FLINK-4911Region-based Task Failover FLIP1/FLINK-4256調(diào)度性能優(yōu)化大規(guī)模批處理作業(yè)優(yōu)化 - JobManagerFlink 自帶的 Batch Shuffle 會(huì)將數(shù)據(jù)托管給 TM混部環(huán)境中，TM被殺是常態(tài)借助 Yarn Auxiliary Service，將數(shù)據(jù)托管給可靠性更高的Node Manager社區(qū)也在嘗試更通用的支持方案：FLIP-31 - Pluggable Shuffle Manager大規(guī)模批處理作業(yè)優(yōu)化 Shuffle Service日志 ETL 作業(yè)雙十一峰值接近1億條每秒手淘主搜索和猜你喜歡樣本量每天數(shù)百 TB實(shí)時(shí)模型直接提升了雙十一效果最大單一批處理作業(yè)處理 400 TB數(shù)據(jù)，最大單節(jié)點(diǎn)并發(fā)20000