Flink 基礎

Flink特性

流式計算是大數據計算的痛點，第1代實時計算引擎Storm對Exactly Once 語義和窗口支持較弱，使用的場景有限且無法支持高吞吐計算；Spark Streaming 采用“微批處理”模擬流計算，在窗口設置很小的場景中有性能瓶頸，Spark 本身也在嘗試連續執行模式（Continuous Processing），但進展緩慢。

Flink是一個低延遲、高吞吐的實時計算引擎，其利用分布式一致性快照實現檢查點容錯機制，并實現了更好的狀態管理，Flink可在毫秒級的延遲下處理上億次/秒的消息或者事件，同時提供了一個Exactly-once的一致性語義，保證了數據的正確性，使得Flink可以提供金融級的數據處理能力，總結其高級特性包括CSTW(CheckPoint，Statue，Time，windows)

 

Flink和Spark對比 

設計思路

Spark的技術理念是基于批來模擬流，微批處理的延時較高（無法優化到秒以下的數量級），且無法支持基于event_time的時間窗口做聚合邏輯。Flink和spark相反，它基于流計算來模擬批計算，更切合數據的生成方式，技術上有更好的擴展性。 

狀態管理

流處理任務要對數據進行統計，如Sum, Count, Min, Max,這些值是需要存儲的，因為要不斷更新，這些值或者變量就可以理解為一種狀態，如果數據源是在讀取Kafka, RocketMQ，可能要記錄讀取到什么位置，并記錄Offset，這些Offset變量都是要計算的狀態。

Flink提供了內置的狀態管理，可以把這些狀態存儲在Flink內部，而不需要把它存儲在外部系統，這樣做的好處：

① 降低了計算引擎對外部系統的依賴以及部署，使運維更加簡單；

② 對性能帶來了極大的提升：如果通過外部去訪問如Redis , HBase 需要網絡及RPC資源，如果通過Flink內部去訪問，只通過自身的進程去訪問這些變量。

同時Flink會定期將這些狀態做Checkpoint持久化，把Checkpoint存儲到一個分布式的持久化系統中，比如HDFS，這樣當Flink的任務出現任何故障時，它都會從最近的一次Checkpoint將整個流的狀態進行恢復，然后繼續運行它的流處理，對用戶沒有任何數據上的影響。

Flink 初探 

設計架構

Flink是一個分層的架構系統，每一層所包含的組件都提供了特定的抽象，用來服務于上層組件，Flink的分層體現有四層，分別是Deploy層、core層、API層/Libraries層，其中Deploy層主要涉及的是Flink的部署模式及同資源調度組件的交互模式，Core層提供了支持Flink計算的全部核心實現，API層/Libraries層提供了Flink的API接口和基于API接口的特定應用的計算框架；

 

Deploy層：該層主要涉及了Flink的部署模式，Flink支持多種部署模式：本地、集群（Standalone/YARN）、云（GCE/EC2），Standalone 部署模式與Spark類似；

Runtime層：Runtime層提供了支持Flink計算的全部核心實現，比如：支持分布式Stream處理、Job Graph到Execution Graph的映射、調度 等，為上層API層提供基礎服務。

API層：API層主要實現了面向無界Stream的流處理和面向Batch的批處理API，其中面向流處理對應DataStream API，面向批處理對應DataSet API。

Libraries層：該層也可以稱為Flink應用框架層，根據API層的劃分，在API層之上構建的滿足特定應用的實時計算框架，也分別對應于面向流處理 和面向批處理兩類。面向流處理支持：CEP（復雜事件處理）、SQL-like的操作（基于Table的關系操作）；面向批處理支持：FlinkML（機器學習庫）、Gelly（圖處理）。 

Flink on yarn

Flink支持增量迭代，具有對迭代自行優化的功能，因此在on yarn上提交的任務性能略好于 Spark，Flink提供2種方式在yarn上提交任務：啟動1個一直運行的 Yarn session（分離模式）和在 Yarn 上運行1個 Flink 任務(客戶端模式)；

分離模式：通過命令yarn-session.sh的啟動方式本質上是在yarn集群上啟動一個flink集群，由yarn預先給flink集群分配若干個container，在yarn的界面上只能看到一個Flink session with X TaskManagers的任務，并且只有一個Flink界面，可以從Yarn的Application Master鏈接進入；

客戶端模式：通過命令bin/flink run -m yarn-cluster啟動，每次發布1個任務，本質上給每個Flink任務啟動了1個集群，yarn在任務發布時啟動JobManager（對應Yarn的AM）和TaskManager，如果一個任務指定了n個TaksManager(-yn n)，則會啟動n+1個Container，其中一個是JobManager，發布m個應用，則有m個Flink界面，不同的任務不可能在一個Container（JVM）中，實現了資源隔離。

進入Flink的bin目錄下運行./yarn-session.sh –help 查看幫助驗證yarn是否成功配置，使用./yarn-session.sh –q 顯示yarn所有nodeManager節點資源；部署On yarn模式的Flink只需要修改配置conf/flink-conf.yaml ，詳細參數請參考官網：通用配置：Configuration，HA配置：High Availability (HA)

采用分離模式來啟動Flink Yarn Session，提交后提示該yarn application成功提交到yarn并返回id，使用yarn application –kill application_id 來停止yarn上提交的任務；

yarn-session.sh -n 3 -jm 700 -tm 700 -s 8 -nm FlinkOnYarnSession -d –st

可以直接提交自帶的詞頻統計用例，驗證on yarn模式是否配置成功：

~/bin/flink run -m yarn-cluster -yn 4 -yjm 2048 -ytm 2048 ~/flink/examples/batch/WordCount.jar  
  

流程分析

分離模式：通過命令yarn-session.sh先啟動集群，然后再提交作業，接著會向yarn申請一塊空間后，資源永遠保持不變。如果資源滿了，下一個作業就無法提交，只能等到yarn中的其中一個作業執行完成后，釋放了資源，下個作業才會正常提交。所有作業共享Dispatcher和ResourceManager；共享資源；適合規模小執行時間短的作業。

DataStream

DataStream是Flink的較低級API，用于進行數據的實時處理任務，可以將該編程模型分為DataSource、Transformation、Sink三個部分；

DataSource

源是程序讀取輸入數據的位置，可以使用 StreamExecutionEnvironment.addSource(sourceFunction) 將源添加到程序，Flink 有許多預先實現的源函數，也可以通過實現 SourceFunction 方法自定義非并行源 ，或通過實現 ParallelSourceFunction 或擴展 RichParallelSourceFunction 自定義并行源。

有幾個預定義的流數據源可從 StreamExecutionEnvironment 訪問：

基于文件：

readTextFile(path) #逐行讀取文本文件（文件符合 TextInputFormat 格式），并作為字符串返回每一行。
 
readFile(fileInputFormat, path) #按指定的文件輸入格式（fileInputFormat）讀取指定路徑的文件。
 
readFile(fileInputFormat, path, watchType, interval, pathFilter) #前兩個方法的內部調用方法。根據給定文件格式（fileInputFormat）讀取指定路徑的文件。根據 watchType，定期監聽路徑下的新數據（FileProcessingMode.PROCESS_CONTINUOUSLY），或者處理當前在路徑中的數據并退出（FileProcessingMode.PROCESS_ONCE），使用 pathFilter，可以進一步排除正在處理的文件。

基于Socket：socketTextStream 從 Socket 讀取，元素可以用分隔符分隔。

基于集合：

fromCollection(Seq) #用 Java.util.Collection 對象創建數據流，集合中的所有元素必須屬于同一類型；
 
fromCollection(Iterator) #用迭代器創建數據流。指定迭代器返回的元素的數據類型；
 
fromElements(elements: _*) #從給定的對象序列創建數據流。所有對象必須屬于同一類型；
 
fromParallelCollection(SplittableIterator) #并行地從迭代器創建數據流。指定迭代器返回的元素的數據類型；
 
generateSequence(from, to) #并行生成給定間隔的數字序列。

自定義：addSource 附加新的源函數。例如從 Apache Kafka 中讀取，可以使用 addSource(new FlinkKafkaConsumer08<>(...))。請詳細查看 連接器。

Transformation

Transformation操作將1個或多個DataStream轉換為新的DataStream，多個轉換組合成復雜的數據流拓撲，如下圖所示，DataStream會由不同的Transformation操作、轉換、過濾、聚合成其他不同的流，從而完成業務要求；

Map：DataStream -> DataStream，一個數據元生成一個新的數據元。將輸入流的元素翻倍:dataStream.map { x => x * 2 }

FlatMap：DataStream -> DataStream，一個數據元生成多個數據元（可以為0）。將句子分割為單詞：

dataStream.flatMap { str => str.split(" ") }

Filter：DataStream -> DataStream，每個數據元執行布爾函數，只保存函數返回 true 的數據元。過濾掉零值的過濾器：

dataStream.filter { _ != 0 }

KeyBy ：DataStream -> KeyedStream，將流劃分為不相交的分區。具有相同 Keys 的所有記錄在同一分區。指定 key 的取值：

dataStream.keyBy("someKey") // Key by field "someKey"
 
dataStream.keyBy(0) // Key by the first element of a Tuple

Reduce ：KeyedStream -> DataStream，KeyedStream 元素滾動執行 Reduce。將當前數據元與最新的一個 Reduce 值組合作為新值發送。創建 key 的值求和：keyedStream.reduce { _ + _ }

Aggregations ：KeyedStream -> DataStream，應用于 KeyedStream 上的滾動聚合。

Window：KeyedStream -> WindowedStream，Windows 可以在已經分區的 KeyedStream 上定義。Windows 根據某些特征（例如，在最近5秒內到達的數據）對每個Keys中的數據進行分組。更多說明參考 Windows 或 譯版。

dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)))

WindowAll ：DataStream -> AllWindowedStream，Windows 也可以在 DataStream 上定義。在許多情況下，這是非并行轉換。所有記錄將收集在 windowAll 算子的一個任務中。

dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))

Window Apply ：WindowedStream -> DataStream 或 AllWindowedStream -> DataStream，將函數應用于整個窗口。一個對窗口數據求和：

windowedStream.apply { WindowFunction }
 
allWindowedStream.apply { AllWindowFunction }

Window Reduce：WindowedStream -> DataStream，Reduce 函數應用于窗口并返回結果值。windowedStream.reduce { _ + _ }

Aggregations on windows：WindowedStream -> DataStream，聚合窗口內容;

Union ：DataStream* -> DataStream，兩個或多個數據流的合并，創建包含來自所有流的所有數據元的新流。如果將數據流與自身聯合，則會在結果流中獲取兩次數據元。

dataStream.union(otherStream1, otherStream2, ...)

Window Join ：DataStream,DataStream -> DataStream，Join 連接兩個流，指定 Key 和窗口。

dataStream.join(otherStream)
 
    .where(<key selector>).equalTo(<key selector>)
 
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
 
    .apply { ... }

Window CoGroup ：DataStream,DataStream -> DataStream，CoGroup 連接兩個流，指定 Key 和窗口。

dataStream.coGroup(otherStream)
 
    .where(0).equalTo(1)
 
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
 
    .apply {}

CoGroup 與 Join 的區別：CoGroup 會輸出未匹配的數據，Join 只輸出匹配的數據

Connect ：DataStream,DataStream -> ConnectedStreams，連接兩個有各自類型的數據流。允許兩個流之間的狀態共享。

someStream : DataStream[Int] = ...
 
otherStream : DataStream[String] = ...
 
val connectedStreams = someStream.connect(otherStream)

可用于數據流關聯配置流；

CoMap, CoFlatMap ：ConnectedStreams -> DataStream，作用域連接數據流（connected data stream）上的 map 和 flatMap：

Split ：DataStream -> SplitStream，將數據流拆分為兩個或更多個流。

Select ：SplitStream -> DataStream，從 SpliteStream 中選擇一個流或多個流。

val even = split select "even"
 
val odd = split select "odd"
 
val all = split.select("even","odd")

Iterate ：DataStream -> IterativeStream -> DataStream，將一個算子的輸出重定向到某個先前的算子，在流中創建 feedback 循環。這對于定義不斷更新模型的算法特別有用。以下代碼以流開頭并連續應用迭代體。大于0的數據元將被發送回 feedback，其余數據元將向下游轉發。

Project：DataStream -> DataStream，作用于元組的轉換，從元組中選擇字段的子集。

DataStream<Tuple3<Integer, Double, String>> in = // [...]
 
DataStream<Tuple2<String, Integer>> out = in.project(2,0); 

Sink

Data Sink 消費 DataStream 并轉發到文件，套接字，外部系統或打印到頁面。Flink 帶有各種內置輸出格式，封裝在 DataStreams 上的算子操作后面：

writeAsText() / TextOutputFormat：按字符串順序寫入文件。通過調用每個元素的 toString() 方法獲得字符串。

writeAsCsv(...) / CsvOutputFormat：將元組寫為逗號分隔的形式寫入文件。行和字段分隔符是可配置的。每個字段的值來自對象的 toString() 方法。

print() / printToErr()：在標準輸出/標準錯誤流上打印每個元素的 toString() 值。可以定義輸出前綴，這有助于區分不同的打印調用。如果并行度大于1，輸出也包含生成輸出的任務的標識符。

writeUsingOutputFormat() / FileOutputFormat：自定義文件輸出的方法和基類。支持自定義對象到字節的轉換。

writeToSocket：將元素寫入 Socket，使用 SerializationSchema 進行序列化。

addSink：調用自定義接收器函數。請詳細查看 連接器。

DataStream 的 write*() 方法主要用于調試目的。他們沒有參與 Flink checkpoint，這意味著這些函數通常具有至少一次的語義。刷新到目標系統的數據取決于 OutputFormat 的實現，并非所有發送到 OutputFormat 的數據都會立即顯示在目標系統中。此外，在失敗的情況下，這些記錄可能會丟失。

要將流可靠、準確地傳送到文件系統，請使用 flink-connector-filesystem。通過 .addSink(...) 方法的自定義實現，可以實現在 checkpoint 中精確一次的語義。

Time

流式數據處理最大的特點是數據具有時間屬性特征，Flink根據時間產生的位置不同，將時間區分為三種概念：數據生成時間（Event_time）、事件接入時間（Ingestion_time）、事件處理時間（Processing_time），用戶可以根據需要選擇事件類型作為流式數據的時間屬性，極大增強了數據處理的靈活性和準確性；

 

Event_time：獨立事件在產生它的設備上的發生時間，這個時間通常在到達Flink之前已經嵌入到生產數據中，因此時間順序取決于事件產生的地方，和下游的數據處理系統的事件無關，需要在Flink中指定事件的時間屬性或者設定時間提取器提取事件時間；

Processing_time：指在操作算子計算過程中獲取到的所在主機的時間，用戶選擇了Processing_time后，所有和時間相關的計算算子都直接使用其所在主機的系統時間，使用Processing_time的程序性能相對較高，延時相對較低，因為其所有操作不需要做任何時間上的對比和協調；

Ingestion_time：指數據接入Flink系統的時間，依賴于Source Operator所在主機的系統時鐘；

一般場景中選擇event_time作為事件時間戳是最貼近生產的，但大多數情況下由于數據的延遲和亂序使用processing_time；

Window窗口

Windows定義和分類

在流式計算中，數據持續不斷的流入計算引擎，需要一個窗口限定計算范圍，比如監控場景的近2分鐘或者精準計算的每隔2分鐘計算一次，窗口定義了該范圍，輔助完成有界范圍的數據處理；

Flink的DataStream API將窗口抽象成獨立的Operator，且支持很多窗口算子，每個窗口算子包含Window Assigner 、Windows Function、觸發器、剔除器、時延設定等部分屬性，其中Window Assigner 和 Windows Function是必須要指定的屬性；

Window Assigner用來決定某個元素被分配到哪個/哪些窗口中去；Trigger觸發器決定了一個窗口何時能夠被計算或清除，每個窗口都會擁有一個自己的Trigger；

Evictor驅逐者在Trigger觸發之后，在窗口被處理之前，Evictor（如果有Evictor的話）會用來剔除窗口中不需要的元素，相當于一個filter。

Flink支持多種窗口類型，按照驅動類型分為：時間驅動的Time Window（如每30秒鐘）和數據驅動的Count Window（如每100個事件），按照窗口的滾動方式又可以分成：翻滾窗口（Tumbling Window，無重疊），滾動窗口（Sliding Window，有重疊）和會話窗口（Session Window，活動間隙），下圖可以看出分類區別：

Time Window 是根據時間對數據流進行分組的，且窗口機制和時間類型是完全解耦的，也就是說當需要改變時間類型時（三種時間）不需要更改窗口邏輯相關的代碼，Time Window 中常見的即為Tumbling Time Window和Sliding Time Window；

Count Window 是根據元素個數對數據流進行分組的，也包括Tumbling Count Window和Sliding Count Window；

Windows實現

每一個窗口都擁有一個屬于自己的 Trigger，Trigger上會有定時器，用來決定一個窗口何時能夠被計算或清除，每當有元素加入到該窗口，或者之前注冊的定時器超時了，那么Trigger都會被調用。Trigger的返回結果可以是 continue（不做任何操作），fire（處理窗口數據），purge（移除窗口和窗口中的數據），或者 fire + purge。一個Trigger的調用結果只是fire的話，那么會計算窗口并保留窗口原樣，也就是說窗口中的數據仍然保留不變，等待下次Trigger fire的時候再次執行計算。一個窗口可以被重復計算多次知道它被 purge 了。在purge之前，窗口會一直占用著內存。

當Trigger fire了，窗口中的元素集合就會交給Evictor（如果指定了的話）。Evictor 主要用來遍歷窗口中的元素列表，并決定最先進入窗口的多少個元素需要被移除。剩余的元素會交給用戶指定的函數進行窗口的計算。如果沒有 Evictor 的話，窗口中的所有元素會一起交給函數進行計算。

計算函數收到了窗口的元素（可能經過了 Evictor 的過濾），并計算出窗口的結果值，并發送給下游。窗口的結果值可以是一個也可以是多個。DataStream API 上可以接收不同類型的計算函數，包括預定義的sum(),min(),max()，還有 ReduceFunction，FoldFunction，還有WindowFunction。WindowFunction 是最通用的計算函數，其他的預定義的函數基本都是基于該函數實現的。

Flink 對于一些聚合類的窗口計算（如sum,min）做了優化，因為聚合類的計算不需要將窗口中的所有數據都保存下來，只需要保存一個result值就可以了。每個進入窗口的元素都會執行一次聚合函數并修改result值。這樣可以大大降低內存的消耗并提升性能。但是如果用戶定義了 Evictor，則不會啟用對聚合窗口的優化，因為 Evictor 需要遍歷窗口中的所有元素，必須要將窗口中所有元素都存下來。

Windows Function

在運用窗口計算時，Flink根據上有數據集是否是KeyedStream類型（數據是否按照Key分區），如果上游數據未分組則調用window()方法指定Windows Assigner，數據會根據Key在不同Task實例中并行計算，最后得出針對每個Key的統計結果，如果是Non-Keyed類型則調用WindowsAll()方法指定Windows Assigner，所有的數據都會在窗口算子中路由得到一個Task中計算，并得到全局統計結果；

定義完窗口分配器后，需要為每一個窗口指定計算邏輯，也就是Windows Function，Flink提供了四種類型Window Function，分別是ReduceFunction、AggreateFunction、FoldFunction、ProcessWindowFunction，其中FoldFunction將逐漸不再使用；四種類型有分為增量聚合操作（ReduceFunction、AggreateFunction、FoldFunction）和全量聚合操作（ProcessWindowFunction）；

增量聚合函數計算性能高，占用存儲空間少，因為其只需要維護窗口的中間結果狀態值，不需要緩存原始數據；全量聚合函數使用代價相對高，性能較弱，因為算子需要緩存該窗口的接入數據，然后等窗口觸發后對所有原始數據進行匯總計算，若接入數據量大或窗口時間長容易導致計算性能下降；

ReduceFunction和AggreateFunction相似，但前者的輸出類型和輸入類型一致（如使用tuple的某個字段聚合），后者更加靈活地提供3個復寫方法，add()定義數據的添加邏輯，getResult()定義根據Accumulator計算結果的邏輯，merge()方法定義合并accumulator的邏輯；

ProcessWindowFunction可以支撐更復雜的算子，其支持基于窗口全部數據元素的結果計算，當算子需要窗口的元數據或狀態數據，或者算子不支持運算交換律和結合律（統計所有元素的中位數和眾數），需要該函數中的Context對象，Context類定義了Window的元數據及可以操作的Window的狀態數據包括GlobalState和WindowState；

大部分情況下，需要增量計算和全量計算結合，因為增量計算雖然一定程度能夠提升窗口性能，但靈活性不及ProcessWindowFunction，兩者整合使用，既可以得到增量算子又可以得到窗口的元數據（窗口開始、終止時間等），比如在計算TOP N的場景中，分窗口計算完數據的計算后需要根據商品ID匯聚總的點擊數；

Watermark

由于網絡或系統等外部因素影響，事件數據不能及時傳輸到Flink系統中，導致數據亂序、延遲等問題，因此需要一種機制能夠控制數據處理的過程和進度；基于event_time時間的Windows創建后，具體如何確定屬于該Windows中的數據元素已經全部到達，如果確定全部到達就可以對所有數據進行窗口計算操作（匯總、分組），如果數據沒有全部到達，則繼續等待該窗口中的數據，但是又不能無限期的等下去，需要有機制來保證一個特定的時間后，必須觸發window去進行計算了，此時watermark發揮作用了，它表示當達到watermark后，在watermark之前的數據已經全部達到(即使后面還有延遲的數據)；Watermark是處理EventTime 窗口計算提出的機制，本質上是一種時間戳，可以在讀取 Source時候指定或者在transformation操作之前，用自定義的Watermark生成器按照需求指定；

正常情況下，流式數據的到達時間是有序的，如下圖：

一般情況存在數據的亂序（out-of-order）和延遲(late element)，此時水位線機制能表明該時間戳之前到當前水位線時間戳的數據已經全部達到，沒有比它（水位線）更早的數據了，并觸發計算；

 

Flink中生成水位線的方式有兩種：Periodic Watermarks（周期性）和Punctuated Watermarks，前者假設當前時間戳減去固定時間，所有數據都能達到，后者要在特定事件指示后觸發生成水位線；

舉例說明Periodic Watermarks 工作方式：當前window為10s，設想理想情況下消息都沒有延遲，那么eventTime等于系統當前時間，假如設置watermark等于eventTime的時候，當watermark = 00:00:10的時候，就會觸發w1的計算，這個時后因為消息都沒有延遲，watermark之前的消息（00:00:00~00:00:10）都已經落入到window中，所以會計算window中全量的數據。那么假如有一條消息eventTime是00:00:01 應該屬于w1，在00:00:11才到達，因為假設消息沒有延遲，那么watermark等于當前時間，00:00:11,這個時候w1已經計算完畢，那么這條消息就會被丟棄，沒有加入計算，這樣就會出現問題。這是已經可以理解，代碼中為什么要減去一個常量作為watermark，假設每次提取eventTime的時減去2s，那么當data1在00:00:11到達的時候，watermark是00:00:09這個時候，w1還沒有觸發計算，那么data1會被加入w1，這個時候計算完全沒有問題，所以減去一個常量是為了對延時的消息進行容錯；

Punctuated Watermarks提供自定義條件生成水位，例如判斷某個數據元素的當前狀態或tuple類型的某個值，如果接入事件中狀態為0則觸發生成watermark，如果狀態不為0則不觸發，需要分別復寫extractTimestamp和checkAndGetNextWatermark方法；

Flink允許提前預定義數據的提取器Timestamp Extractors，在讀取source時候定義提取時間戳；

延遲數據

基于Event_time的窗口計算雖然可以使用warterMark機制容忍部分延遲，但只能一定程度的緩解該問題，無法應對某些延遲特別嚴重的場景。Flink默認丟失延遲數據，但用戶可以自定義延遲數據的處理方式，此時需要Allowed Lateness機制近數據的額外處理；

DataStream API提供Allowed Lateness方法指定是否對遲到數據進行處理，參數是Time類型的時間間隔大小，代表允許的最大延遲時間，Flink的窗口計算中會將Window的Endtime加上該時間作為窗口最后釋放的結束時間（P），當接入的數據中Event time未超過該時間（P），但WaterMark已經超過Window的Event_Time時直接觸發窗口計算，若Event_Time超過了時間P，則做丟棄處理；

通常情況下可以使用sideOutputLateData 方法對遲到數據進行標記，然后使用getSideOutput()方法得到被標記的延遲數據，分析延遲原因；

多流合并/關聯 

合并

Connect：Flink 提供connect方法實現兩個流或多個流的合并，合并后生成ConnectedStreams，會對兩個流的數據應用不同的處理方法，并且雙流之間可以共享狀態(比如計數)；ConnectedStream提供的map()和flatMap()需要定義CoMapFunction和CoFlatMapFunction分別處理輸入的DataStream數據集；

Union：Union算子主要實現兩個或者多個輸入流合并成一個數據集，需要保證兩個流的格式一致，輸出的流與輸入完全一致；

關聯

Flink支持窗口的多流關聯，即在一個窗口上按照相同條件對多個輸入流進行join操作，需要保證輸入的Stream構建在相同的Windows上，且有相同類型的Key做為關聯條件；

數據集inputStream1通過join方法形成JoinedStreams類型數據集，調用where()方法指定inputStream1數據集的key，調用equalTo()方法指定inputStream2對應關聯的key，通過window()方法指定Window Assigner，最后通過apply()方法中傳入用戶自定義的JoinFunction或者FlatJoinFunction對輸入數據元素進行窗口計算；

Windows Join過程中所有的Join操作都是Inner Join類型，也就是必須滿足相同窗口中，每個Stream都有Key，且key相同才能完成關聯操作并輸出結果； 

狀態和容錯

有狀態計算是Flink重要特性，其內部存儲計算產生的中間結果并提供給后續的Function或算子使用，狀態數據維系在本地存儲中，可以是Flink的堆內存或者堆外內存中，也可以借助于第三方的存儲介質，同storm+ redis / hbase模式相比，Flink完善的狀態管理減少了對外部系統的依賴，減少維護成本；

State和類型

Flink根據數據集是否根據key分區將狀態分為Keyed State和 Operator State兩種類型，Keyed State只能用于KeyedStream類型數據集對應的Function和Operation上，它是Operator State的特例；

Operator State只和并行的算子實例綁定，和數據元素中的key無關，支持當算子實例并行度發生變化后自動重新分配狀態數據；

Keyed State和 Operator State均有兩種形式，一種是托管狀態，一種是原始狀態，前者有Flink Runtime控制和管理狀態數據并將狀態數據轉換成內存Hash tables 或RocksDB的對象存儲，后者由算子自己管理數據結構，當觸發CheckPoint后，Flink并不知道狀態數據內部的數據結構，只是將數據轉換成bytes數據存儲在CheckPoint中，當從Checkpoint恢復任務時，算子自己反序列化出狀態的數據結構；

CheckPoint 和SavePoint

Flink基于輕量級分布式快照算法提供了CheckPoint機制，分布式快照可以將同一時間點的Task/Operator狀態數據全局統一快照處理，包括Keyed State和Operator State

Savepoints是檢查點的一種特殊實現，底層使用CheckPoint機制，Savepoint是用戶以手工命令方式觸發CheckPoint，并將結果持久化到指定的存儲路徑中，其主要目的是幫助用戶在升級和維護集群過程中保存系統的狀態數據，避免因停機運維或者升級到知道正常終止的應用數據狀態無法恢復。