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