一、flink基礎(chǔ)核心知識(shí)1、什么是Flink2014年Flink作為主攻流計(jì)算的大數(shù)據(jù)引擎開始在開源大數(shù)據(jù)行業(yè)內(nèi)嶄露頭角。區(qū)別于Storm、Spark Streaming 以及其他流式計(jì)算引擎的是它不僅是一個(gè)高吞吐、低延遲的 計(jì)算引擎同時(shí)還提供很多高級(jí)功能。比如它提供有狀態(tài)的計(jì)算支持狀態(tài)管理支持強(qiáng)一致性的數(shù)據(jù)語(yǔ)義以及支持Event Time,WaterMark 對(duì)消息亂序的處理等。2015年是流計(jì)算百花齊放的時(shí)代各個(gè)流計(jì)算框架層出不窮。Storm, JStorm, Heron,Flink, Spark StreamingGoogle Dataflow(后來(lái)的Beam)等等。其中Flink的一致性語(yǔ)義和最接近Dataflow模型的開源實(shí)現(xiàn)使其成為流計(jì)算框架中最耀眼的一顆。也許這也是阿里看中Flink的原因并決心投入重金去研究基于Flink的Blink框架。2、Flink的優(yōu)點(diǎn)1、支持批處理和數(shù)據(jù)流程序處理2、優(yōu)雅流暢的支持java和scala api3、同時(shí)支持高吞吐量和低延遲4、支持事件處理和無(wú)序處理通過(guò)SataStream API基于DataFlow數(shù)據(jù)流模型5、在不同的時(shí)間語(yǔ)義(事件時(shí)間攝取時(shí)間、處理時(shí)間)下支持靈活的窗口(時(shí)間滑動(dòng)、翻滾會(huì)話自定義觸發(fā)器)6、僅處理一次的容錯(cuò)擔(dān)保7、自動(dòng)反壓機(jī)制8、圖處理(批) 機(jī)器學(xué)習(xí)(批) 復(fù)雜事件處理(流)9、在dataSet(批處理)API中內(nèi)置支持迭代程序(BSP)10、高效的自定義內(nèi)存管理和健壯的切換能力在in-memory和out-of-core中11、兼容hadoop的mapreduce和storm12、集成YARN,HDFS,Hbase 和其它hadoop生態(tài)系統(tǒng)的組件3、flink 與 sparkStreaming的區(qū)別1、計(jì)算模型flink流計(jì)算spark微批處理2、時(shí)間語(yǔ)義flink事件時(shí)間、處理時(shí)間spark處理時(shí)間3、窗口flink多靈活spark少不靈活窗口必須是批次的整數(shù)倍4、狀態(tài)flink有spark沒(méi)有5、流式SQLflink有spark沒(méi)有二、flink系統(tǒng)架構(gòu)1、JobManager主節(jié)點(diǎn)集群大腦負(fù)責(zé)任務(wù)調(diào)度、資源管理、Checkpoint 協(xié)調(diào)、客戶端接入1、 內(nèi)嵌 ResourceManager管理整個(gè)集群的 Slot 資源2、可配置 HA多 JobManager 主備避免單點(diǎn)故障。2、TaskManager從節(jié)點(diǎn)工作節(jié)點(diǎn)獨(dú)立 JVM 進(jìn)程負(fù)責(zé)實(shí)際執(zhí)行計(jì)算任務(wù)1、每個(gè) TaskManager 管理固定數(shù)量的 Slot預(yù)配置2、啟動(dòng)后主動(dòng)向 JobManager 注冊(cè)領(lǐng)取任務(wù)。3、客戶端提交作業(yè)的節(jié)點(diǎn)負(fù)責(zé)生成 JobGraph 并提交給 JobManager。4、運(yùn)行流程集群?jiǎn)?dòng)先啟動(dòng) JobManager再啟動(dòng)所有 TaskManagerTaskManager 主動(dòng)向 JobManager 注冊(cè)匯報(bào) Slot 資源作業(yè)提交客戶端執(zhí)行flink run生成 JobGraph 提交給 JobManager調(diào)度執(zhí)行JobManager 根據(jù)作業(yè)并行度申請(qǐng) Slot將 Task 分發(fā)到對(duì)應(yīng) TaskManager任務(wù)執(zhí)行TaskManager 啟動(dòng)線程執(zhí)行 Task通過(guò)網(wǎng)絡(luò)交互數(shù)據(jù)定期匯報(bào)狀態(tài)。5、并行度1、定義同一個(gè)邏輯算子被拆分成獨(dú)立的子任務(wù)SubTask同時(shí)運(yùn)行的個(gè)數(shù)為并行度每個(gè)子任務(wù)處理一部分?jǐn)?shù)據(jù)。一個(gè)流程序的并行度是其所有算子中最大的并行度。一個(gè)程序中不同算子可能有不同的并行度。2、并行度的 4 個(gè)設(shè)置層級(jí)優(yōu)先級(jí)從高到低① 算子級(jí)最高優(yōu)先級(jí)在代碼中針對(duì)單個(gè)算子單獨(dú)設(shè)置并行度粒度最細(xì)僅作用于當(dāng)前算子。// 給 map 算子設(shè)置并行度為 4dataStream.map(x-x*2).setParallelism(4)② 執(zhí)行環(huán)境級(jí)針對(duì)整個(gè)作業(yè)的所有算子設(shè)置默認(rèn)并行度算子級(jí)未設(shè)置時(shí)生效StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(6);// 整個(gè)作業(yè)所有算子默認(rèn)并行度為 6③ 客戶端提交級(jí)提交作業(yè)時(shí)通過(guò)命令行參數(shù)指定./bin/flink run-m JobManager地址:8081-p 8-c 全類名 jar包路徑④ 集群配置級(jí)最低優(yōu)先級(jí)Flink 集群配置文件flink-conf.yaml中的全局默認(rèn)值所有作業(yè)未指定時(shí)生效。parallelism.default: 16、算子鏈Operator Chain1、定義算子鏈?zhǔn)?Flink 的執(zhí)行層優(yōu)化機(jī)制在滿足條件的前提下把多個(gè)相鄰的算子合并成一個(gè)算子鏈作為一個(gè)獨(dú)立的執(zhí)行任務(wù)Task在同一個(gè)線程中運(yùn)行。算子鏈的并行度 鏈內(nèi)所有算子的并行度2、優(yōu)點(diǎn)① 減少線程切換開銷多個(gè)算子在同一個(gè)線程執(zhí)行避免線程間切換② 減少網(wǎng)絡(luò)傳輸同一算子鏈內(nèi)的數(shù)據(jù)直接在內(nèi)存中傳遞不需要序列化 / 反序列化和網(wǎng)絡(luò)通信③ 降低延遲、提升吞吐量少了跨線程 / 跨節(jié)點(diǎn)的交互。3、算子鏈的形成條件同時(shí)滿足以下 4 個(gè)條件并行度相同上下游算子的并行度必須一致Forward 分區(qū)器上下游之間是一對(duì)一的數(shù)據(jù)轉(zhuǎn)發(fā)forward分區(qū)沒(méi)有 shuffle、沒(méi)有數(shù)據(jù)重分區(qū)未禁用算子鏈代碼中沒(méi)有手動(dòng)禁用算子鏈優(yōu)化同一個(gè) Slot 共享組上下游算子屬于同一個(gè) Slot 共享組默認(rèn)都屬于default組。4、自定義算子鏈①.disableChaining()徹底禁用當(dāng)前算子的算子鏈前后都斷開env.是全局禁用也可以單個(gè)算子禁用②.startNewChain()從當(dāng)前算子開始開啟一條新的算子鏈和上游斷開。7、任務(wù)槽Task Slot1、定義任務(wù)槽Task Slot是TaskManager 中資源分配的最小單位每個(gè) Slot 對(duì)應(yīng) TaskManager 的一部分內(nèi)存資源。2、特點(diǎn)①內(nèi)存隔離每個(gè) Slot 擁有獨(dú)立的內(nèi)存空間不同 Slot 之間的內(nèi)存不共享避免任務(wù)互相搶占內(nèi)存②CPU 共享Slot 只隔離內(nèi)存不隔離 CPUCPU 所有 Slot 共享3、Slot 的數(shù)量與配置每個(gè) TaskManager 的 Slot 數(shù)量是在配置文件flink-conf.yaml里修改參數(shù)taskmanager.numberOfTaskSlots:1靜態(tài)配置的啟動(dòng)后固定不變生產(chǎn)環(huán)境通常設(shè)置為和 TaskManager 的 CPU 核心數(shù)一致4、Slot 共享機(jī)制①什么是 Slot 共享同一個(gè)作業(yè)的不同Task不同算子/算子鏈可以在同一個(gè) Slot執(zhí)行前提是它們必須屬于同一個(gè) Slot 共享組默認(rèn)所有算子都屬于default共享組自動(dòng)開啟共享可以通過(guò).slotSharingGroup(group1)手動(dòng)指定分組不同分組之間不能共享 Slot。也就是說(shuō)一個(gè) Slot 里可以同時(shí)運(yùn)行多個(gè)不同算子/算子鏈的 Task比如 source 算子的 Task 和keyBy 算子的 Task 可以放在同一個(gè) Slot 里同時(shí)運(yùn)行。②Slot 共享的優(yōu)點(diǎn)提高資源利用率避免輕量算子獨(dú)占資源造成浪費(fèi)負(fù)載均衡Task 自動(dòng)均勻分布到各個(gè) Slot避免部分節(jié)點(diǎn)過(guò)載。三、flink部署模式1、本地模式Local Mode純開發(fā)調(diào)試模式所有 Flink 角色都運(yùn)行在同一個(gè) JVM 進(jìn)程中用多線程模擬分布式并行不具備真實(shí)分布式能力僅用于驗(yàn)證代碼邏輯、單元測(cè)試。在 IDEA 里直接運(yùn)行 Flink 代碼本質(zhì)就是本地模式。2、獨(dú)立集群模式Standalone ModeFlink 自帶的獨(dú)立分布式集群模式不依賴任何外部資源調(diào)度框架Flink 自己管理集群資源是最基礎(chǔ)的分布式部署形態(tài)。如果資源不足或出現(xiàn)故障沒(méi)有自動(dòng)擴(kuò)展或重分配資源的保證必須手動(dòng)處理。所以一般只用在測(cè)試或作業(yè)很少的場(chǎng)景1會(huì)話模式1、原理需要先啟動(dòng)一個(gè)集群保持一個(gè)會(huì)話在這個(gè)會(huì)話中通過(guò)客戶端提交作業(yè)集群?jiǎn)?dòng)時(shí)所有資源就都已經(jīng)確定所有提交的作業(yè)會(huì)競(jìng)爭(zhēng)集群中的資源。比較適用于單個(gè)規(guī)模小執(zhí)行時(shí)間短的大量作業(yè)。#啟動(dòng)集群命令./bin/start-cluster.sh#命令行提交作業(yè)./bin/flink run-m JobManager地址:8081-c 全類名 jar包路徑#停止集群命令./bin/stop-cluster.sh2應(yīng)用模式1、原理會(huì)話模式代碼都是在客戶端上執(zhí)行然后由客戶端提交給JobManager的。應(yīng)用模式直接把應(yīng)用提交到JobManager上運(yùn)行。也就代表我們需要為每一個(gè)提交的應(yīng)用單獨(dú)啟動(dòng)一個(gè)JobManager集群。#應(yīng)用程序的jar包必須放在lib/目錄下#執(zhí)行以下命令啟動(dòng)JobManager./bin/standalone-job.shstart--job-classname 全類名#自己?jiǎn)?dòng)TaskManager,需要幾個(gè)啟動(dòng)幾個(gè)./bin/taskmanager.shstart#停止集群./bin/taskmanager.sh stop./bin/standalone-job.sh stop3、YARN 部署模式Flink 把資源調(diào)度完全交給 YARN按需申請(qǐng)資源和 Hadoop 生態(tài)深度打通是目前大廠最主流的部署方式1會(huì)話模式1、基于yarn原理客戶端把flink應(yīng)用提交給yarn的ResourceManager,yarn的ResourceManager會(huì)向yarn的NodeManager申請(qǐng)容器在容器上flink會(huì)部署JobManager和TaskManager的實(shí)例從而啟動(dòng)集群。flink會(huì)根據(jù)運(yùn)行在JobManager上的作業(yè)所需要的slot數(shù)量動(dòng)態(tài)分配TaskManager資源。所有作業(yè)都提交到這個(gè)共享集群里運(yùn)行#啟動(dòng)一個(gè)session集群,對(duì)yarn來(lái)講就是啟動(dòng)了一個(gè)應(yīng)用./bin/yarn-session.sh-d-nm sessionTest#命令行提交任務(wù)./bin/flink run-d-c 全類名 jar包路徑#會(huì)自動(dòng)提交到啟動(dòng)session集群因?yàn)?tmp/.yarn-properties文件會(huì)記錄集群的ip端口直接連接#命令行集群關(guān)閉echostop|./bin/yarn-session.sh-id 應(yīng)用id(application-開頭的2單作業(yè)模式會(huì)話模式因?yàn)橘Y源共享會(huì)導(dǎo)致很多問(wèn)題所以為了更好的隔離資源考慮為每個(gè)作業(yè)啟動(dòng)一個(gè)集群就是單作業(yè)模式作業(yè)完成后集群會(huì)關(guān)閉所有資源也會(huì)釋放。但flink本身無(wú)法直接這樣運(yùn)行一般需要借助一些資源管理框架來(lái)啟動(dòng)集群比如yarnk8s.#命令行提交任務(wù)./bin/flink run-d-t yarn-per-job-c 全類名 jar包路徑#停止作業(yè)./bin/flink list-t yarn-per-job-Dyarn.application.id application開頭的id#獲取對(duì)應(yīng)的jobid./bin/flink cancel-t yarn-per-job-Dyarn.application.id application開頭的id 剛獲取的jobid3應(yīng)用模式前面兩種模式下應(yīng)用代碼都是在客戶端上執(zhí)行然后由客戶端提交給JobManager的但是這樣客戶端需要占用大量網(wǎng)絡(luò)帶寬去下載依賴和把二進(jìn)制數(shù)據(jù)發(fā)送給JobManager很多情況下我們提交作業(yè)用的是同一個(gè)客戶端就會(huì)加重客戶端所在節(jié)點(diǎn)的資源消耗。所以我們不要客戶端了直接把應(yīng)用提交到JobManager上運(yùn)行。也就代表我們需要為每一個(gè)提交的應(yīng)用單獨(dú)啟動(dòng)一個(gè)JobManager集群。#命令行提交任務(wù)./bin/flink run-application-t yarn-application-c 全類名 jar包路徑#停止作業(yè)./bin/flink list-t yarn-application-Dyarn.application.id application開頭的id#獲取對(duì)應(yīng)的jobid./bin/flink cancel-t yarn-application-Dyarn.application.id application開頭的id 剛獲取的jobid四、批處理與流處理實(shí)例1、批處理publicclassWordCountBath{publicstaticvoidmain(String[]args)throwsException{//創(chuàng)建執(zhí)行環(huán)境ExecutionEnvironmentenvExecutionEnvironment.getExecutionEnvironment();//讀取數(shù)據(jù)DataSourceStringlinedataenv.readTextFile(文件路徑);//切分轉(zhuǎn)換數(shù)據(jù) (flink word ,轉(zhuǎn)換成flink,1word,1FlatMapOperatorString,Tuple2String,IntegerwordAndOnelinedata.flatMap(newFlatMapFunctionString,Tuple2String,Integer(){OverridepublicvoidflatMap(Stringvalue,CollectorTuple2String,Integerout)throwsException{//按照空格切分單詞String[]wordsvalue.split( );//將單詞轉(zhuǎn)換為flink,1)for(Stringword:words){Tuple2String,IntegerwordTuple2Tuple2.of(word,1);//使用Collector向下游發(fā)送數(shù)據(jù)out.collect(wordTuple2);}}});//按照word分組UnsortedGroupingTuple2String,IntegerwordAndOneGroupBywordAndOne.groupBy(0);//分組內(nèi)聚合AggregateOperatorTuple2String,IntegersumwordAndOneGroupBy.sum(1);//1是位置表示第二個(gè)元素//輸出sum.print();}}2、流處理publicclassWordCountStream{publicstaticvoidmain(String[]args)throwsException{//創(chuàng)建執(zhí)行環(huán)境StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();//讀取數(shù)據(jù)DataStreamSourceStringlinedataenv.readTextFile(文件路徑);//有界流DataStreamSourceStringsocketdataenv.socketTextStream(hadoop102,8888);//無(wú)界流//切分轉(zhuǎn)換數(shù)據(jù) (flink word ,轉(zhuǎn)換成flink,1word,1SingleOutputStreamOperatorTuple2String,IntegerwordAndOnelinedata.flatMap(newFlatMapFunctionString,Tuple2String,Integer(){OverridepublicvoidflatMap(Stringvalue,CollectorTuple2String,Integerout)throwsException{//按照空格切分單詞String[]wordsvalue.split( );//將單詞轉(zhuǎn)換為flink,1)for(Stringword:words){Tuple2String,IntegerwordTuple2Tuple2.of(word,1);//使用Collector向下游發(fā)送數(shù)據(jù)out.collect(wordTuple2);}}});//按照word分組KeyedStreamTuple2String,Integer,StringwordAndOneKeyBywordAndOne.keyBy(newKeySelectorTuple2String,Integer,String(){OverridepublicStringgetKey(Tuple2String,Integervalue)throwsException{returnvalue.f0;}});//分組內(nèi)聚合SingleOutputStreamOperatorTuple2String,IntegersumwordAndOneKeyBy.sum(1);//1是位置表示第二個(gè)元素//輸出sum.print();//執(zhí)行:類似spark streaming最后的.start()env.execute();}}