Flink概念模型1. 核心概念Streams（流（liú）），流分為有界流和無界流有（yǒu）界流指的是有（yǒu）固定大小，不隨（suí）時間增加而增長的數據，比如www.17C.com保存在 Hive 中的一個表；而無界流指的是（shì）數據隨著時間增加而增（zēng）長，計算（suàn）狀態持續進行（háng），比如www.17C.com消費 Kafka 中的消息，消息持續不斷，那麽計算也會持續進行不會結束。

    S展示沙盤tate（狀（zhuàng）態（tài）），所謂的狀態指的是在進（jìn）行流（liú）式計算過程中的信息一般用作（zuò）容錯恢複和持久化，流式計算（suàn）在本質上是（shì）增量計算，也就是（shì）說需要（yào）不斷地查詢過去的狀態狀（zhuàng）態在（zài） Flink 中有十分重要的作用，例如為了確保 Exactly-once 語義需要將數據寫到狀態中（zhōng）；此外，狀態的持久（jiǔ）化存儲也是集群出現 Fail-o展示沙盤ver 的情況下自動（dòng）重啟的前提條件。

    Time（時間），Flink 支持了 Event time、Ingestion time、Processing time 等多種時間語義，時間是www.17C.com在進行 Flink 程序開發時判斷（duàn）業務狀態是否滯（zhì）後和延遲的重要依據。

    API：Flink 自身提供了不同級別的抽象來支展示沙盤持www.17C.com（men）開發流（liú）式或者批量處理程序，由上而下可分為 SQL / Table API、DataStream API、ProcessFunction 三層，開發者可以根（gēn）據需要選擇不同層級的 API 進行開發。

    2. 編程模型Flink 程序的基礎構建模塊是流（Streams）和轉換（Transformatio展示沙盤（pán）ns），每一個數據（jù）流起始於一（yī）個或多個 Source，並終止於一個或多個 Sink數據流類似於有向（xiàng）無環圖。

    （DAG）3. Flink 集群模型和角色在實（shí）際生產中（zhōng），Flink 都是以集群在運行，在運行的過程中包含了兩類進程JobManager：它扮演的是集群管理者的角色，負（fù）責（zé）調（diào）度任務、協調 checkp展示沙盤oints、協調故障恢複、收集 Job 的狀態信息，並管理 Flink 集群中的從（cóng）節點 TaskManager。

    TaskManager：實際負責（zé）執行計算的 Worker，在其上執行（háng） Flink Job 的一組 Task；TaskManager 還是所在節點的管理（lǐ）員，它負責把（bǎ）該節點上的服務器信息比如內展示沙盤存（cún）、磁盤、任務（wù）運行情況等向 JobManager 匯報。

    Client：用戶（hù）在提交編寫好的 Flink 工程時，會先創建一個客戶端再進（jìn）行提交，這個客戶端就是 Client，Client 會根據用戶傳入的參數選擇使用 yarn per job 模式、stand-alone 模式還（hái）是 yarn-sessi展示（shì）沙盤on 模式將 Flink 程序提交到集群。

    4. Flink 資源和資源組Flink 集群（qún）中，一個 TaskManger 就是一個 JVM 進程，並且會用獨立（lì）的線程來執行 task，為了控製一個 TaskManger 能接受多少（shǎo）個 task，可以（yǐ）簡單的把 Task Slot 理解為（wéi） TaskManag展示沙盤er 的計算資（zī）源子集。

    假如一個 TaskManager 擁有 5 個 slot，那麽該 TaskManager 的計算資源會被平均分為（wéi） 5 份，不同的 task 在（zài）不同的 slot 中執行，避免（miǎn）資源競爭（zhēng）但是需要（yào）注意的是，slot 僅僅用來做內存的隔離，對 CPU 不起作用。

    那麽（me）運行在同一個 JVM 展示沙盤的 task 可以共（gòng）享 TCP 連接，減少網絡傳輸，在一定程度上（shàng）提高了程序的運行效率，降低了資源消耗與此同時，Flink 還允許將不能形成算子鏈（liàn）的兩個操作，比如下圖（tú）中的 flatmap 和 key&sink 放（fàng）在（zài）一個 TaskSlot 裏執行以達到（dào）資源共享的（de）目的。

    5. 架構&Flink四（sì）層圖結構Fl展示（shì）沙盤ink 也采用了經典的主從模式，DataFlow Graph 與 Storm 形成的拓撲 Topology 結構類似，Flink 程序啟動後，會根據用戶的代碼處理成 Stream Graph，然後優化成為 JobGraph，JobManager 會根據 JobGraph 生成 ExecutionGr展示沙盤（pán）aph。

    ExecutionGraph 才（cái）是 Flink 真正能執行的數據結構，當很多個 ExecutionGraph 分（fèn）布在集群（qún）中（zhōng），就會形成一張網狀的拓撲結構Flink 中的執行（háng）圖可以分成四層：StreamGraph -> JobGraph -> ExecutionGraph -> 物理（lǐ）執行圖。

    l 展示沙盤（pán）StreamGraph：是根據用戶通過（guò） Stream API 編寫的代碼（mǎ）生成的最初的圖用來表示程序的拓撲結構l JobGraph：StreamGraph經過優化後（hòu）生成（chéng）了（le） JobGraph，提交給 JobManager 的數據結構。

    主要的優化為，將多個符合條件的節（jiē）點 chain 在（zài）一起作為一個節點，這展示沙盤樣可以減少數據在節點之間流動（dòng）所需要（yào）的序列化/反序列化/傳輸消耗l ExecutionGraph：JobManager 根據 JobGraph 生成ExecutionGraph。

    ExecutionGraph是JobGraph的並行化版本，是調度層最核心的（de）數據結（jié）構l 物理執行圖（tú）：JobManager 根展（zhǎn）示（shì）沙盤據 ExecutionGraph 對 Job 進行調度後，在各個TaskManager 上部署 Task 後形成的“圖”，並不是一個具體的數據結構。

    StreamGraph：根據用戶通過 Stream API 編寫的代碼生成的最初的圖l StreamNode：用來代表 operator 的類，並具有所展示沙盤有相關的屬性，如並發度、入邊和出邊等l StreamEdge：表示連接兩個StreamNode的（de）邊。

    JobGraph：StreamGraph經過優（yōu）化後生成了 JobGraph，提交給 JobManager 的數據（jù）結構l JobVertex：經過（guò）優化後符合（hé）條件的多個StreamNode可能會chain展示沙盤在一起生成一個JobVertex，即一個JobVertex包含一個或多個operator，JobVertex的輸入是JobEdge，輸出是IntermediateDataSet。

    l IntermediateDataSet：表示JobVertex的輸出，即經過operator處理產生的數據集produ展示沙盤cer是JobVertex，consumer是JobEdgel JobEdge：代表了job graph中的一條數據傳輸通道。

    source 是 IntermediateDataSet，target 是 JobVertex即（jí）數據通過JobEdge由IntermediateDataSet傳（chuán）遞給目標Job展示沙盤VertexExecutionGraph

    ：JobManager 根據 JobGraph 生成ExecutionGraphExecutionGraph是JobGraph的並行化版本，是調度層最核心的數據結構l ExecutionJobVertex：和JobGraph中的JobVertex一（yī）一對應。

    每一展示沙盤個ExecutionJobVertex都有（yǒu）和並發度一樣多的 ExecutionVertexl ExecutionVertex：表示ExecutionJobVertex的其中（zhōng）一個並發（fā）子任務，輸入是ExecutionEdge，輸出是IntermediateResultPartition。

    l Interm展示沙盤ediateResult：和JobGraph中的IntermediateDataSet一一對應一個IntermediateResult包含多個IntermediateResultPartition，其個數等於該operator的並發度。

    l IntermediateResultPartition：表示展示沙盤ExecutionVertex的（de）一個輸（shū）出分區，producer是（shì）ExecutionVertex，consumer是若幹（gàn）個ExecutionEdge。

    l ExecutionEdge：表（biǎo）示ExecutionVertex的輸入，source是IntermediateResultPartition，targ展示沙盤et是ExecutionVertexsource和target都隻能是一個。

    l Execution：是執行（háng）一個 ExecutionVertex 的一次嚐試當發生故障或者（zhě）數據需要重算的情況下 ExecutionVertex 可能會有多個（gè） ExecutionAttemptID一（yī）個 Execution 通展示沙盤過（guò） ExecutionAttemptID 來唯一標識。

    JM和（hé）TM之間（jiān）關於 task 的部署和 task status 的更新都是通（tōng）過（guò） ExecutionAttemptID 來確定消息接受者 物理（lǐ）執行圖：JobManager 根據 ExecutionGraph 對 Job 進行調度後，在各個TaskM展示沙盤anager 上部署 Task 後形成的“圖”，並不是一（yī）個具體的（de）數據結構（gòu）。

    l Task：Execution被調度後在分配的 TaskManager 中啟動對應（yīng）的 TaskTask 包裹了具有用戶執行邏輯的 operatorl ResultPartition：代表由一個Task的生成的（de）數據，和Exec展示沙盤utionGraph中的IntermediateResultPartition一一對應。

    l ResultSubpartition：是ResultPartition的一個子（zǐ）分區每（měi）個ResultPartition包含多（duō）個ResultSubpartition，其數目要由下遊消費 Task 數和 Distr展示（shì）沙盤（pán）ibutionPattern 來決定。

    l InputGate：代表Task的輸入封裝（zhuāng），和JobGraph中JobEdge一一對應每個InputGate消費了一個或多個的ResultPartitionl InputChannel：每個InputGate會包含一個以上（shàng）的InputChannel，和Exe展示沙盤cutionGraph中的ExecutionEdge一一對應，也和ResultSubpartition一（yī）對一地相（xiàng）連，即一個InputChannel接收一個ResultSubpartition的輸出。

    6. 反壓（BackPressure）反壓是分布式處理係（xì）統中（zhōng）經常遇到的問題，當消費（fèi）者速度低於生（shēng）產者的速展示沙盤度（dù）時，則需要消費者將信息反饋給（gěi）生產者使得生產（chǎn）者（zhě）的速度能和消（xiāo）費者（zhě）的速度進（jìn）行（háng）匹配7. 時間的種類針對stream數據中的（de）時間，可以（yǐ）分為以下三種：。

    Ø Event Time：事件產生的時間，它（tā）通（tōng）常（cháng）由事件（jiàn）中的時間（jiān）戳描述Ø Ingestion time：事件（日（rì）誌，數據，消息）進入Flink的（de）時間(不考慮)Ø展示（shì）沙盤 Processing Time：事（shì）件被處理時當前係統的時間。

    8. 容錯Flink 基於兩階段提交實現了（le）精確的一次處（chù）理語（yǔ）義Storm 在容錯（cuò）方麵隻支持了 Record 級別的 ACK-FAIL，發送出去的每（měi）一條消息，都可以確定是被成功處理或失敗處理，因此 Storm 支持至少（shǎo）處理一次語義。

    9. 分布展示沙盤（pán）式緩存分布式緩存最初（chū）的思想誕生於 Hadoop 框架，Hadoop 會將一些數據（jù）或者（zhě）文件緩存在 HDFS 上，在分布式環境中讓所有的計算節點（diǎn）調（diào）用同（tóng）一個（gè）配置文（wén）件（jiàn）在 Flink 中，Flink 框架開發者們同樣將這個特性進行了實現。

    Flink 提供的分布式緩存類型 Hadoop，目的是為了在分布式環境中展示沙盤讓每一個 TaskManager 節點保存一份相同（tóng）的數據或者文件，當前計（jì）算節點的 task 就像讀取本地文件一樣拉取這些配置如（rú）何使用分布式緩存，使用分（fèn）布式緩存有兩個步驟。

    第一步：首先需（xū）要在 env 環境中注（zhù）冊一個文件，該文件可以來源於本地，也可（kě）以來源於 HDFS ，並（bìng）且為該（gāi）文件取一個（gè）名字第二步：在使展示沙（shā）盤用分布式緩存時，可根據注（zhù）冊的名字（zì）直（zhí）接獲取10. 狀態所謂的狀態，其實指的是 Flink 程序的中間計算結果。

    Flink 支持了（le）不同類型的狀態，並且針對狀態的持久化還提供了（le）專門的機製和狀態管理器When working with state, it might also be useful to re展示（shì）沙盤ad about Flink’s state backends. Flink provides different state backends that specify how and where state is stored. State can be located on Java’s he展示沙盤ap or off-heap. Depending on your state backend, Flink can also manage the state for the application, meaning Flink deals with the memory management (展示沙盤possibly spilling to disk if necessary) to allow applications to hold very large state. State backends can be configured without changing your applica展示沙盤tion logic.

    Flink 的官網同樣給（gěi）出了適用於狀態計算的（de）幾種情況：l When an application searches for certain event patterns, the state will store the sequence of events encounter展示沙盤ed so far

    l When aggregating events per minute/hour/day, the state holds the pending aggregatesl When training a machine learning model over a stream o展示沙盤f data points, the state holds the current version of the model parameters

    l When historic data needs to be managed, the state allows efficient access 展示沙盤to events that occurred in the past

    以上四種情（qíng）況分別是：複雜事件處理獲取符合某一特定時間（jiān）規則的事（shì）件、聚合計算（suàn）、機器學習的模型訓練、使用曆史的數據進行計算10.1 狀態的類型Flink中有兩種基本類（lèi）型的State：Keyed State，Operator State，展示沙盤他們兩種都可（kě）以以兩種形式存（cún）在（zài）：原始狀態(raw state)和托管狀態（tài）(managed state)。

    托管狀態：由Flink框架管理的狀態，www.17C.com（men）通常使用的就是這種原始狀（zhuàng）態：由用戶自行管理狀態具體的數據結構，框架（jià）在做checkpoint的時候，使用byte[]來讀寫狀態內容，對其內部數據結構一無所知通展示（shì）沙盤常在DataStream上（shàng）的（de）狀（zhuàng）態推薦使用托管的狀態（tài），當實現一個用戶自定（dìng）義的operator時，會使用到原始（shǐ）狀態。

    但是www.17C.com工作中一般不常用，所以www.17C.com不考慮他Operator Statel operator state是task級別的state，說白了就是每（měi）個task對應一個（gè）statel Kafka C展示沙盤onnector source中的每個分區（task）都需要記錄（lù）消費的topic的partition和offset等信息。

    l operator state 隻有（yǒu）一種托管狀態：ValueStateKeyed Statel keyed state 記錄的是每個key的狀態l Keyed state托管狀展示沙盤態有六種類型：ValueState

    ListStateMapStateReducingStateAggregatingStateFoldingState10.2 狀態後端種類和（hé）配置默認情況下，Flink 的狀（zhuàng）態會保存在 taskmanager 的內存中，Flink 提供了三種可用的狀態（tài）後端用於在不同情展示沙盤況（kuàng）下（xià）進行狀態後端的保存。

    MemoryStateBackendFsStateBackendRocksDBStateBackendMemoryStateBackend默認情況（kuàng）下，狀態信息（xī）是存（cún）儲在 TaskManager 的堆內存中的，checkpoint 的時（shí）候將（jiāng）狀態保存到JobManager 的堆內存展示沙盤中。

    MemoryStateBackend 適用於www.17C.com本地調試使用，來（lái）記錄一些狀態很小（xiǎo）的（de） Job 狀態信息每個獨立的狀態（state）默認限製大小為（wéi） 5MB，可以通（tōng）過構造函數增加容量狀態的大小不能超過（guò） akka 的 Framesize 大小

    聚合後的狀態必須能夠放進 JobManager 的內存中缺點：展示沙盤隻能保存數據量（liàng）小的狀態狀態數據有可能會丟失優點：開發測試很方便FsStateBackend狀態信息存（cún）儲在 TaskManager 的堆內存中的，checkpoint 的（de）時候將狀態（tài）保存到（dào）指定的文（wén）件中 (HDFS等文（wén）件係統)

    缺點：狀態大小受TaskManager內存限（xiàn）製(默認支（zhī）持5M)優（yōu）點：狀態訪問速度展示沙盤很快（kuài）狀（zhuàng）態信息不（bú）會丟失用於： 生產，也可存儲狀態數（shù）據量大的情（qíng）況RocksDBStateBackend狀態信息存儲在 RocksDB 數（shù）據庫 (key-value 的數據（jù）存儲服務)， 最終保存在本地文件中

    checkpoint 的時候將狀態保存到指定的文件中 (HDFS 等（děng）文件係統)與（yǔ） FsStateBa展示沙盤（pán）ckend 不同的（de）是，RocksDBStateBackend 將正在運（yùn）行中的狀態數（shù）據保存在 RocksDB 數據庫中，RocksDB 數據庫默認將數據存儲在 TaskManager 運行節點的數據目錄下。

    這意味著（zhe），RocksDBStateBackend 可以存儲遠超過 FsStateBackend 展示沙盤的狀態，可以（yǐ）避免向 FsStateBackend 那樣一旦出現狀態暴增會導（dǎo）致 OOM，但是因為將狀（zhuàng）態數據保存在 RocksDB 數據庫中（zhōng），吞吐量會有（yǒu）所下降。

    缺點：狀態訪問速度有所下降優點：可以存（cún）儲超大量（liàng）的狀態信息狀態信（xìn）息不會（huì）丟失用於：也適用於大作業、狀態較大、全局高可用的那些任務10.3 State展示沙盤backend配置方式（1）單任務調整修改當前任務代碼env.setStateBackend(new

        FsStateBackend("hdfs://namenode:9000/flink/checkpoints"));

        或（huò）者new MemoryStateBackend展示沙盤()

        或者new RocksDBStateBackend(filebackend, true);【需要添加第三（sān）方依賴】

    （2）全局調整修改flink-conf.yamlstate.backend: filesystem

        state.checkpoints.dir: hd展示沙（shā）盤fs://namenode:9000/flink/checkpoints

        注意：state.backend的值可（kě）以是下麵幾種：jobmanager(MemoryStateBackend),

        filesystem(FsStateBackend), rocksdb(Roc展示沙盤ksDBStateBackend)

    11. checkpoint 機製11.1 checkpoint概述（1）為了保證state的容錯性，Flink需要對state進行checkpoint（2）Checkpoint是Flink實現容錯機製最核心的功能（néng），它能夠根據配置周期性地基於Stream中各個。

    Op展示沙（shā）盤erator/task的（de）狀態來生（shēng）成快照，從而將這些狀態數據定期持（chí）久化存儲下來，當Flink程序一旦意外崩潰時（shí），重新運行程序（xù）時（shí）可以有選（xuǎn）擇地從這些快照進行恢複，從（cóng）而修正因為故障帶來的程序數據異常（3）Flink的checkpoint機製可以與(stream和state)的持久化存儲交互的（de）前提：

    持久化的s展示沙盤ource，它需（xū）要支持在一定時間內重放事（shì）件這種（zhǒng）sources的典型例子是持久化的消息隊列（liè）（比如Apache Kafka，RabbitMQ等）或文件係統（比如HDFS，S3，GFS等（děng））用於state的（de）持久化存儲，例如分布式文件（jiàn）係統（比如HDFS，S3，GFS等）。

    11.2 checkpoint配置默認展示沙盤checkpoint功能是disabled的（de），想要使用的時候需要先啟用，checkpoint開啟之後，checkPointMode有兩種，Exactly-once和At-least-once，默認（rèn）的checkPointMode是Exactly-once，

    Exactly-once對（duì）於（yú）大多（duō）數應用來說是最展（zhǎn）示（shì）沙盤（pán）合適的At-least-once可（kě）能（néng）用在某些延遲超低的應用程序（始終延遲（chí）為（wéi）幾毫秒）默認checkpoint功能是disabled的，想要使用的時候需要先啟（qǐ）用

        StreamExecutionEnvironment env =

        StreamExecutionEnviron展示沙盤ment.getExecutionEnvironment();

        // 每隔1000 ms進行啟動（dòng）一個檢查點【設置checkpoint的周期】

        env.enableCheckpointing(1000);

        // 高級選項：

        // 設置模式展（zhǎn）示沙盤為exactly-once （這（zhè）是默認值）

        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 確保檢查點之間有至少500 ms的間隔【checkpoin展（zhǎn）示沙盤t最小（xiǎo）間（jiān）隔】

        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

        // 檢（jiǎn）查點必須在一分鍾內完成（chéng），或者被丟棄【checkpoint的超時時間】

        env.getCheckpointC展示沙盤onfig().setCheckpointTimeout(60000);

        // 同一時間隻允許進行一個檢查點

        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        // 表示一旦Flink處展示沙盤（pán）理程序被（bèi）cancel後，會保留Checkpoint數據，以便根據實際需要恢複到指定的Checkpoint【詳細（xì）解釋見備注（zhù）】

        env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCl展示沙盤eanup.RETAIN_ON_CANCELLATION);。

    12. 故障恢（huī）複和重啟策略（luè）12.1 故（gù）障恢複Flink 支持了不同級別的故障恢複策（cè）略，jobmanager.execution.failover-strategy 的可配置項（xiàng）有兩種：full 和（hé） region。

    Full: 集（jí）群中的 Tas展示沙（shā）盤k 發生故障，那麽該任務的所有 Task 都會發生重啟Region: Flink 會把（bǎ）www.17C.com的任務分成不同的 Region，當某一個 Task 發生故障（zhàng）時，Flink 會計算需要（yào）故障（zhàng）恢（huī）複的最（zuì）小 Region。

    Flink 在判斷需要重啟的 Region 時，采（cǎi）用了以下的判斷邏輯：發生錯誤的（de） Task 所展示沙盤（pán）在的 Region 需要重啟（qǐ）；如果（guǒ）當前 Region 的依賴數據出現損壞或者部分丟失，那麽生產數據的（de） Region 也需要重啟；

    為了保（bǎo）證數據一致性，當前 Region 的下遊 Region 也需要（yào）重啟12.2 重啟策略Flink支持不同的重啟策略，以在故障發生時控製作（zuò）業如何（hé）重（chóng）啟，集群在啟（qǐ）動時會伴隨一展示沙盤個默認的重啟策（cè）略，在沒有定義具體重啟策（cè）略時會使用（yòng）該默認策（cè）略。

     如果在工作提（tí）交（jiāo）時指定了一個重啟策略，該策略會覆蓋集群（qún）的默認（rèn）策略，默認的重啟策略可以通過 Flink 的配置文件 flink-conf.yaml 指定配置參數restart-strategy 定義了哪個策略被使用。

    常用的重（chóng）啟策（cè）略（1）固定（dìng）間隔展示沙盤 (Fixed delay)（2）失敗率 (Failure rate)（3）無重啟 (No restart)如果沒有啟（qǐ）用 checkpointing，則使用無重啟 (no restart) 策略。

    如果啟用了 checkpointing，但沒有配（pèi）置重啟策略，則使用固定間隔 (fixed-delay) 展示沙（shā）盤策略（luè）， 嚐試重啟次數默認值是：Integer.MAX_VALUE，重啟策略可以在flink-conf.yaml中配置，表示全局的配置。

    也（yě）可以在應用代碼中動態指定，會覆蓋（gài）全局配置12.3 重（chóng）啟策略配置固定間隔（gé） (Fixed delay)第（dì）一（yī）種：全局配置 flink-conf.yamlrestart-s展示沙盤trategy: fixed-delay

        restart-strategy.fixed-delay.attempts: 3

        restart-strategy.fixed-delay.delay: 10 s

    第二種：應用代碼設置env.setRestartStrategy展示沙盤(RestartStrategies.fixedDelayRestart(

        3, // 嚐試重啟的次數

        Time.of(10, TimeUnit.SECONDS) // 間隔

        ));

    失敗率 (Failure rate)第一種：全局配置 flink-co展示沙盤nf.yamlrestart-strategy: failure-rate

        restart-strategy.failure-rate.max-failures-per-interval: 3

        restart-strategy.failure-rate.failure展示（shì）沙盤-rate-interval: 5 min

        restart-strategy.failure-rate.delay: 10 s

    第二種：應（yīng）用代碼設置env.setRestartStrategy(RestartStrategies.failureRateRestart(

        展示沙盤3, // 一個時間段內的最大失敗次數

        Time.of(5, TimeUnit.MINUTES), // 衡量（liàng）失敗次數的是時間段

        Time.of(10, TimeUnit.SECONDS) // 間隔

        ));

    無重啟第一種：全局配置 flink-conf展示沙盤.yamlrestart-strategy: none第二種：應用代碼設置env.setRestartStrategy(RestartStrategies.noRestart());

    12.4 多checkpoint默認情況下，如果設置了Checkpoint選項，則Flink隻保留最近成功生成的1個C展示沙盤heckpoint，而（ér）當Flink程序失（shī）敗時，可以從最近的這個Checkpoint來進行恢複。

    但是，如（rú）果www.17C.com希望保留多個Checkpoint，並能夠根據實際需（xū）要選擇其中一個（gè）進行恢複，這樣會更加靈活（huó），比如，我（wǒ）們發現最近（jìn）4個小時（shí）數據記錄處理有問題，希（xī）望將整個狀態還原到4小時之前Flink可以（yǐ）支持保留多個展（zhǎn）示沙盤Checkpoint，需要在Flink的配置文件conf/flink-conf.yaml中（zhōng），添加如下（xià）配置，指定最多需要保存Checkpoint的個數（shù）：。

    state.checkpoints.num-retained: 2013. Window13.1 Window概述聚（jù）合事件（比（bǐ）如計數、求和）在流上的展示沙盤工作方式與批處理不同比如（rú），對流中的所有元素進行計數是不可能的，因為通常流是無限的（無界的）。

    所以，流上的聚合需要由（yóu） window 來劃定範圍，比如 “計算過去的5分鍾” ，或者 “最後100個元素的（de）和” window是一種可以把無限數據切割為有限數據塊的手段窗口（kǒu）可以是 時間驅（qū）動的 【Time Win展示沙盤（pán）dow】（比如：每30秒）或者 數據（jù）驅動（dòng）的【Count Window】。

    （比如：每（měi）100個元素）13.2 Window類型聚合事件（比如計數、求和）在流上的工作方式與批處理不同比如（rú），對流中的所（suǒ）有元素進行計數是不可能的，因為通常流是無限（xiàn）的（無界的（de））所以，流上（shàng）的（de）聚合需要由 window 來（lái）劃定範圍，比如展示沙盤 “計（jì）算過去的5分鍾” ，或（huò）者 “最後100個元素的和” 。

    window是一種可以把無限數據切割為（wéi）有限數據塊（kuài）的手段窗口可以是 時間驅動的 【Time Window】（比（bǐ）如：每30秒）或者（zhě） 數據驅動的【Count Window】（比如：每100個元素）窗口通常被區分為不同的類型:

    tumbling wi展示沙盤ndows：滾動窗（chuāng）口 【沒有重疊】 （time，count）sliding windows：滑動窗口 【有重疊（dié）】（time，count）session windows：會話窗口（kǒu） (time)

    global windows: 沒有窗（chuāng）口Tumblingwindows：滾動窗口【沒有重疊】sliding w展示沙盤indowSession windowGlobal Window13.3 Window 操作（zuò）

    Keyed Window & Non-Keyed Window14. CEP複雜事件處理Complex Event Processing（CEP）是 Flink 提供的一個非常亮眼的功能，CEP, is ev展示沙盤ent processing that combines data from multiple sources to infer events or patterns that suggest more complicated circumstances. The goal of complex e展示沙盤vent processing is to identify meaningful events (such as opportunities or threats) and respond to them as quickly as possible.

    在www.17C.com的（de）實（shí）際生產中，隨著數據的實時性要求越來越展示沙盤高（gāo），實時數據（jù）的量也在不斷膨脹，在某些業務場景中需要根（gēn）據連續的實時數據，發現（xiàn）其中有價值的那些事件14.1 程序結構Flink CEP 的程序結構主要分為兩個步驟：。

    定（dìng）義（yì）模式匹配結果第一步，定義一個模式 Pattern，在這裏定義了（le）一個這樣的模（mó）式（shì），即在所有接（jiē）收到的事件中匹配（pèi）那些以 id 等於 42 的事展示沙盤件，然後匹配 volume 大於 10.0 的事件，繼續匹配一個 name 等於 end 的事件；

    第二步，匹配模式並且發出報警，根據定（dìng）義的 pattern 在輸入（rù）流上進行匹配，一旦命中www.17C.com的（de）模（mó）式，就發出（chū）一個（gè）報警14.2 模式定義Flink 支持了非常豐富的模式定義，這些模式也是（shì）www.17C.com（men）實現（xiàn）複雜業務邏輯的展示沙盤基礎。