Flink 從入門到實踐：初步介紹

Apache Flink 是一個框架和分布式處理引擎，用于在 無邊界 和 有邊界 數(shù)據(jù)流上進行 有狀態(tài) 的計算。Flink 能在所有常見集群環(huán)境中運行，并能以內存速度和任意規(guī)模進行計算。

1.架構

1.1 處理無界和有界數(shù)據(jù)

任何類型的數(shù)據(jù)都可以形成一種事件流。信用卡交易、傳感器測量、機器日志、網(wǎng)站或移動應用程序上的用戶交互記錄，所有這些數(shù)據(jù)都形成一種流。

數(shù)據(jù)可以被作為 無界 或者 有界 流來處理。

• 無界流：有定義流的開始，但沒有定義流的結束。它們會無休止地產生數(shù)據(jù)。無界流的數(shù)據(jù)必須持續(xù)處理，即數(shù)據(jù)被攝取后需要立刻處理。我們不能等到所有數(shù)據(jù)都到達再處理，因為輸入是無限的，在任何時候輸入都不會完成。處理無界數(shù)據(jù)通常要求以特定順序攝取事件，例如事件發(fā)生的順序，以便能夠推斷結果的完整性。
• 有界流：有定義流的開始，也有定義流的結束。有界流可以在攝取所有數(shù)據(jù)后再進行計算。有界流所有數(shù)據(jù)可以被排序，所以并不需要有序攝取。有界流處理通常被稱為批處理

Apache Flink 擅長處理無界和有界數(shù)據(jù)集。精確的時間控制和狀態(tài)化使得 Flink 的運行時（runtime）能夠運行任何處理無界流的應用。有界流則由一些專為固定大小數(shù)據(jù)集特殊設計的算法和數(shù)據(jù)結構進行內部處理，產生了出色的性能。

1.2 部署應用到任意地方

Apache Flink 是一個分布式系統(tǒng)，它需要計算資源來執(zhí)行應用程序。Flink 集成了所有常見的集群資源管理器，例如 Hadoop YARN、 Apache Mesos 和 Kubernetes，但同時也可以作為獨立集群運行。

Flink 被設計為能夠很好地工作在上述每個資源管理器中，這是通過資源管理器特定（resource-manager-specific）的部署模式實現(xiàn)的。Flink 可以采用與當前資源管理器相適應的方式進行交互。

部署 Flink 應用程序時，F(xiàn)link 會根據(jù)應用程序配置的并行性自動標識所需的資源，并從資源管理器請求這些資源。在發(fā)生故障的情況下，F(xiàn)link 通過請求新資源來替換發(fā)生故障的容器。提交或控制應用程序的所有通信都是通過 REST 調用進行的，這可以簡化 Flink 與各種環(huán)境中的集成。

1.3 運行任意規(guī)模應用

Flink 旨在任意規(guī)模上運行有狀態(tài)流式應用。因此，應用程序被并行化為可能數(shù)千個任務，這些任務分布在集群中并發(fā)執(zhí)行。所以應用程序能夠充分利用無盡的 CPU、內存、磁盤和網(wǎng)絡 IO。而且 Flink 很容易維護非常大的應用程序狀態(tài)。其異步和增量的檢查點算法對處理延遲產生最小的影響，同時保證精確一次狀態(tài)的一致性。

Flink 用戶報告了其生產環(huán)境中一些令人印象深刻的擴展性數(shù)字

• 處理每天數(shù)萬億的事件
• 應用維護幾TB大小的狀態(tài)
• 應用在數(shù)千個內核上運行

1.4 利用內存性能

有狀態(tài)的 Flink 程序針對本地狀態(tài)訪問進行了優(yōu)化。任務的狀態(tài)始終保留在內存中，如果狀態(tài)大小超過可用內存，則會保存在能高效訪問的磁盤數(shù)據(jù)結構中。任務通過訪問本地（通常在內存中）狀態(tài)來進行所有的計算，從而產生非常低的處理延遲。Flink 通過定期和異步地對本地狀態(tài)進行持久化存儲來保證故障場景下精確一次的狀態(tài)一致性。

2.應用

Apache Flink 是一個針對無界和有界數(shù)據(jù)流進行有狀態(tài)計算的框架。Flink 自底向上在不同的抽象級別提供了多種 API，并且針對常見的使用場景開發(fā)了專用的擴展庫。

2.1 流處理應用的基本組件

2.1.1 流

顯而易見，（數(shù)據(jù)）流是流處理的基本要素。然而，流也擁有著多種特征。這些特征決定了流如何以及何時被處理。Flink 是一個能夠處理任何類型數(shù)據(jù)流的強大處理框架。

• 有界 和 無界 的數(shù)據(jù)流：流可以是無界的；也可以是有界的，例如固定大小的數(shù)據(jù)集。Flink 在無界的數(shù)據(jù)流處理上擁有諸多功能強大的特性，同時也針對有界的數(shù)據(jù)流開發(fā)了專用的高效算子。
• 實時 和 歷史記錄 的數(shù)據(jù)流：所有的數(shù)據(jù)都是以流的方式產生，但用戶通常會使用兩種截然不同的方法處理數(shù)據(jù)?；蚴窃跀?shù)據(jù)生成時進行實時的處理；亦或是先將數(shù)據(jù)流持久化到存儲系統(tǒng)中——例如文件系統(tǒng)或對象存儲，然后再進行批處理。Flink 的應用能夠同時支持處理實時以及歷史記錄數(shù)據(jù)流。

2.1.2 狀態(tài)

只有在每一個單獨的事件上進行轉換操作的應用才不需要狀態(tài)，換言之，每一個具有一定復雜度的流處理應用都是有狀態(tài)的。任何運行基本業(yè)務邏輯的流處理應用都需要在一定時間內存儲所接收的事件或中間結果，以供后續(xù)的某個時間點（例如收到下一個事件或者經(jīng)過一段特定時間）進行訪問并進行后續(xù)處理。

應用狀態(tài)是 Flink 中的一等公民，F(xiàn)link 提供了許多狀態(tài)管理相關的特性支持，其中包括：

• 多種狀態(tài)基礎類型：Flink 為多種不同的數(shù)據(jù)結構提供了相對應的狀態(tài)基礎類型，例如原子值（value），列表（list）以及映射（map）。開發(fā)者可以基于處理函數(shù)對狀態(tài)的訪問方式，選擇最高效、最適合的狀態(tài)基礎類型。
• 插件化的 State Backend：State Backend 負責管理應用程序狀態(tài)，并在需要的時候進行 checkpoint。Flink 支持多種 state backend，可以將狀態(tài)存在內存或者 RocksDB。RocksDB 是一種高效的嵌入式、持久化鍵值存儲引擎。Flink 也支持插件式的自定義 state backend 進行狀態(tài)存儲。
• 精確一次語義：Flink 的 checkpoint 和故障恢復算法保證了故障發(fā)生后應用狀態(tài)的一致性。因此，F(xiàn)link 能夠在應用程序發(fā)生故障時，對應用程序透明，不造成正確性的影響。
• 超大數(shù)據(jù)量狀態(tài)：Flink 能夠利用其異步以及增量式的 checkpoint 算法，存儲數(shù) TB 級別的應用狀態(tài)。
• 可彈性伸縮的應用：Flink 能夠通過在更多或更少的工作節(jié)點上對狀態(tài)進行重新分布，支持有狀態(tài)應用的分布式的橫向伸縮。

2.1.3 時間

時間是流處理應用另一個重要的組成部分。因為事件總是在特定時間點發(fā)生，所以大多數(shù)的事件流都擁有事件本身所固有的時間語義。進一步而言，許多常見的流計算都基于時間語義，例如窗口聚合、會話計算、模式檢測和基于時間的 join。流處理的一個重要方面是應用程序如何衡量時間，即區(qū)分事件時間（event-time）和處理時間（processing-time）。

Flink 提供了豐富的時間語義支持。

• 事件時間模式：使用事件時間語義的流處理應用根據(jù)事件本身自帶的時間戳進行結果的計算。因此，無論處理的是歷史記錄的事件還是實時的事件，事件時間模式的處理總能保證結果的準確性和一致性。
• Watermark 支持：Flink 引入了 Watermark 的概念，用以衡量事件時間進展。Watermark 也是一種平衡處理延時和完整性的靈活機制。
• 遲到數(shù)據(jù)處理：當以帶有 Watermark 的事件時間模式處理數(shù)據(jù)流時，在計算完成之后仍會有相關數(shù)據(jù)到達。這樣的事件被稱為遲到事件。Flink 提供了多種處理遲到數(shù)據(jù)的選項，例如將這些數(shù)據(jù)重定向到旁路輸出（side output）或者更新之前完成計算的結果。
• 處理時間模式：除了事件時間模式，F(xiàn)link 還支持處理時間語義。處理時間模式根據(jù)處理引擎的機器時鐘觸發(fā)計算，一般適用于有著嚴格的低延遲需求，并且能夠容忍近似結果的流處理應用。

2.2 分層 API

Flink 根據(jù)抽象程度分層，提供了三種不同的 API。每一種 API 在簡潔性和表達力上有著不同的側重，并且針對不同的應用場景。

2.2.1 ProcessFunction

ProcessFunction 是 Flink 所提供的最具表達力的接口。ProcessFunction 可以處理一或兩條輸入數(shù)據(jù)流中的單個事件或者歸入一個特定窗口內的多個事件。它提供了對于時間和狀態(tài)的細粒度控制。開發(fā)者可以在其中任意地修改狀態(tài)，也能夠注冊定時器用以在未來的某一時刻觸發(fā)回調函數(shù)。因此，你可以利用 ProcessFunction 實現(xiàn)許多有狀態(tài)的事件驅動應用所需要的基于單個事件的復雜業(yè)務邏輯。

下面的代碼示例展示了如何在 KeyedStream 上利用 KeyedProcessFunction 對標記為 START 和 END 的事件進行處理。當收到 START 事件時，處理函數(shù)會記錄其時間戳，并且注冊一個時長 $4$ 小時的計時器。如果在計時器結束之前收到 END 事件，處理函數(shù)會計算其與上一個 START 事件的時間間隔，清空狀態(tài)并將計算結果返回。否則，計時器結束，并清空狀態(tài)。

/**
* Matches keyed START and END events and computes the difference between
* both elements' timestamps. The first String field is the key attribute,
* the second String attribute marks START and END events.
*/
public static class StartEndDuration
  extends KeyedProcessFunction<String, Tuple2<String, String>, Tuple2<String, Long>> {

  private ValueState<Long> startTime;
  
  @Override
  public void open(Configuration conf) {
    // obtain state handle
    startTime = getRuntimeContext()
      .getState(new ValueStateDescriptor<Long>("startTime", Long.class));
  }

  /** Called for each processed event. */
  @Override
  public void processElement(
      Tuple2<String, String> in,
      Context ctx,
      Collector<Tuple2<String, Long>> out) throws Exception {
  
      switch (in.f1) {
        case "START":
          // set the start time if we receive a start event.
          startTime.update(ctx.timestamp());
          // register a timer in four hours from the start event.
          ctx.timerService()
            .registerEventTimeTimer(ctx.timestamp() + 4 * 60 * 60 * 1000);
          break;
        case "END":
          // emit the duration between start and end event
          Long sTime = startTime.value();
          if (sTime != null) {
            out.collect(Tuple2.of(in.f0, ctx.timestamp() - sTime));
            // clear the state
            startTime.clear();
          }
        default:
          // do nothing
      }
  }

  /** Called when a timer fires. */
  @Override
  public void onTimer(
    long timestamp,
    OnTimerContext ctx,
    Collector<Tuple2<String, Long>> out) {

    // Timeout interval exceeded. Cleaning up the state.
    startTime.clear();
  }
}

2.2.2 DataStream API

DataStream API 為許多通用的流處理操作提供了處理原語。這些操作包括窗口、逐條記錄的轉換操作，在處理事件時進行外部數(shù)據(jù)庫查詢等。DataStream API 支持 Java 和 Scala 語言，預先定義了例如 map()、reduce()、aggregate() 等函數(shù)。你可以通過擴展實現(xiàn)預定義接口或使用 Java、Scala 的 lambda 表達式實現(xiàn)自定義的函數(shù)。

下面的代碼示例展示了如何捕獲會話時間范圍內所有的點擊流事件，并對每一次會話的點擊量進行計數(shù)。

// a stream of website clicks
DataStream<Click> clicks = ...

DataStream<Tuple2<String, Long>> result = clicks
  // project clicks to userId and add a 1 for counting
  .map(
    // define function by implementing the MapFunction interface.
    new MapFunction<Click, Tuple2<String, Long>>() {
      @Override
      public Tuple2<String, Long> map(Click click) {
        return Tuple2.of(click.userId, 1L);
      }
    })
  // key by userId (field 0)
  .keyBy(0)
  // define session window with 30 minute gap
  .window(EventTimeSessionWindows.withGap(Time.minutes(30L)))
  // count clicks per session. Define function as lambda function.
  .reduce((a, b) -> Tuple2.of(a.f0, a.f1 + b.f1));

2.2.3 SQL & Table API

Flink 支持兩種關系型的 API，Table API 和 SQL。這兩個 API 都是批處理和流處理統(tǒng)一的 API，這意味著在無邊界的實時數(shù)據(jù)流和有邊界的歷史記錄數(shù)據(jù)流上，關系型 API 會以相同的語義執(zhí)行查詢，并產生相同的結果。Table API 和 SQL 借助了 Apache Calcite 來進行查詢的解析，校驗以及優(yōu)化。它們可以與 DataStream 和 DataSet API 無縫集成，并支持用戶自定義的標量函數(shù)，聚合函數(shù)以及表值函數(shù)。

Flink 的關系型 API 旨在簡化數(shù)據(jù)分析、數(shù)據(jù)流水線和 ETL 應用的定義。

下面的代碼示例展示了如何使用 SQL 語句查詢捕獲會話時間范圍內所有的點擊流事件，并對每一次會話的點擊量進行計數(shù)。此示例與上述 DataStream API 中的示例有著相同的邏輯。

SELECT userId, COUNT(*)
FROM clicks
GROUP BY SESSION(clicktime, INTERVAL '30' MINUTE), userId

2.3 庫

Flink 具有數(shù)個適用于常見數(shù)據(jù)處理應用場景的擴展庫。這些庫通常嵌入在 API 中，且并不完全獨立于其它 API。它們也因此可以受益于 API 的所有特性，并與其他庫集成。

• 復雜事件處理（CEP）：模式檢測是事件流處理中的一個非常常見的用例。Flink 的 CEP 庫提供了 API，使用戶能夠以例如正則表達式或狀態(tài)機的方式指定事件模式。CEP 庫與 Flink 的 DataStream API 集成，以便在 DataStream 上評估模式。CEP 庫的應用包括 網(wǎng)絡入侵檢測，業(yè)務流程監(jiān)控 和 欺詐檢測。
• DataSet API：DataSet API 是 Flink 用于批處理應用程序的核心 API。DataSet API 所提供的基礎算子包括 map、reduce、(outer) join、co-group、iterate 等。所有算子都有相應的算法和數(shù)據(jù)結構支持，對內存中的序列化數(shù)據(jù)進行操作。如果數(shù)據(jù)大小超過預留內存，則過量數(shù)據(jù)將存儲到磁盤。Flink 的 DataSet API 的數(shù)據(jù)處理算法借鑒了傳統(tǒng)數(shù)據(jù)庫算法的實現(xiàn)，例如混合散列連接（hybrid hash-join）和外部歸并排序（external merge-sort）。
• Gelly：Gelly 是一個可擴展的圖形處理和分析庫。Gelly 是在 DataSet API 之上實現(xiàn)的，并與 DataSet API 集成。因此，它能夠受益于其可擴展且健壯的操作符。Gelly 提供了內置算法 ，如 label propagation、triangle enumeration 和 page rank 算法，也提供了一個簡化自定義圖算法實現(xiàn)的 Graph API 。

3.運維

Apache Flink 是一個針對無界和有界數(shù)據(jù)流進行有狀態(tài)計算的框架。由于許多流應用程序旨在以最短的停機時間連續(xù)運行，因此流處理器必須提供出色的故障恢復能力，以及在應用程序運行期間進行監(jiān)控和維護的工具。

Apache Flink 非常注重流數(shù)據(jù)處理的可運維性。接下來，我們將詳細介紹 Flink 的故障恢復機制，并介紹其管理和監(jiān)控應用的功能。

3.1 7 × 24小時穩(wěn)定運行

在分布式系統(tǒng)中，服務故障是常有的事，為了保證服務能夠 $7?24$ 小時穩(wěn)定運行，像 Flink 這樣的流處理器故障恢復機制是必須要有的。顯然這就意味著，它（這類流處理器）不僅要能在服務出現(xiàn)故障時候能夠重啟服務，而且還要當故障發(fā)生時，保證能夠持久化服務內部各個組件的當前狀態(tài)，只有這樣才能保證在故障恢復時候，服務能夠繼續(xù)正常運行，好像故障就沒有發(fā)生過一樣。

Flink 通過幾下多種機制維護應用可持續(xù)運行及其一致性:

• 檢查點的一致性：Flink 的故障恢復機制是通過建立分布式應用服務狀態(tài)一致性檢查點實現(xiàn)的，當有故障產生時，應用服務會重啟后，再重新加載上一次成功備份的狀態(tài)檢查點信息。結合可重放的數(shù)據(jù)源，該特性可保證 精確一次（exactly-once） 的狀態(tài)一致性。
• 高效的檢查點：如果一個應用要維護一個 TB 級的狀態(tài)信息，對此應用的狀態(tài)建立檢查點服務的資源開銷是很高的，為了減小因檢查點服務對應用的延遲性（SLAs 服務等級協(xié)議）的影響，F(xiàn)link 采用異步及增量的方式構建檢查點服務。
• 端到端的精確一次：Flink 為某些特定的存儲支持了事務型輸出的功能，及時在發(fā)生故障的情況下，也能夠保證精確一次的輸出。
• 集成多種集群管理服務：Flink 已與多種集群管理服務緊密集成，如 Hadoop YARN，Mesos，以及 Kubernetes。當集群中某個流程任務失敗后，一個新的流程服務會自動啟動并替代它繼續(xù)執(zhí)行。
• 內置高可用服務：Flink 內置了為解決單點故障問題的高可用性服務模塊，此模塊是基于 Apache ZooKeeper 技術實現(xiàn)的，Apache ZooKeeper 是一種可靠的、交互式的、分布式協(xié)調服務組件。

3.2 Flink 能夠更方便地升級、遷移、暫停、恢復應用服務

驅動關鍵業(yè)務服務的流應用是經(jīng)常需要維護的。比如需要修復系統(tǒng)漏洞，改進功能，或開發(fā)新功能。然而升級一個有狀態(tài)的流應用并不是簡單的事情，因為在我們?yōu)榱松壱粋€改進后版本而簡單停止當前流應用并重啟時，我們還不能丟失掉當前流應用的所處于的狀態(tài)信息。

而 Flink 的 Savepoint 服務就是為解決升級服務過程中記錄流應用狀態(tài)信息及其相關難題而產生的一種唯一的、強大的組件。一個 Savepoint，就是一個應用服務狀態(tài)的一致性快照，因此其與 checkpoint 組件的很相似，但是與 checkpoint 相比，Savepoint 需要手動觸發(fā)啟動，而且當流應用服務停止時，它并不會自動刪除。Savepoint 常被應用于啟動一個已含有狀態(tài)的流服務，并初始化其（備份時）狀態(tài)。Savepoint 有以下特點：

• 便于升級應用服務版本：Savepoint 常在應用版本升級時使用，當前應用的新版本更新升級時，可以根據(jù)上一個版本程序記錄的 Savepoint 內的服務狀態(tài)信息來重啟服務。它也可能會使用更早的 Savepoint 還原點來重啟服務，以便于修復由于有缺陷的程序版本導致的不正確的程序運行結果。
• 方便集群服務移植：通過使用 Savepoint，流服務應用可以自由的在不同集群中遷移部署。
• 方便 Flink 版本升級：通過使用 Savepoint，可以使應用服務在升級 Flink 時，更加安全便捷。
• 增加應用并行服務的擴展性：Savepoint 也常在增加或減少應用服務集群的并行度時使用。
• 便于 A/B 測試及假設分析場景對比結果：通過把同一應用在使用不同版本的應用程序，基于同一個 Savepoint 還原點啟動服務時，可以測試對比 $2$ 個或多個版本程序的性能及服務質量。
• 暫停和恢復服務：一個應用服務可以在新建一個 Savepoint 后再停止服務，以便于后面任何時間點再根據(jù)這個實時刷新的 Savepoint 還原點進行恢復服務。
• 歸檔服務：Savepoint 還提供還原點的歸檔服務，以便于用戶能夠指定時間點的 Savepoint 的服務數(shù)據(jù)進行重置應用服務的狀態(tài)，進行恢復服務。

3.3 監(jiān)控和控制應用服務

如其它應用服務一樣，持續(xù)運行的流應用服務也需要監(jiān)控及集成到一些基礎設施資源管理服務中，例如一個組件的監(jiān)控服務及日志服務等。監(jiān)控服務有助于預測問題并提前做出反應，日志服務提供日志記錄能夠幫助追蹤、調查、分析故障發(fā)生的根本原因。最后，便捷易用的訪問控制應用服務運行的接口也是 Flink 的一個重要的亮點特征。

Flink 與許多常見的日志記錄和監(jiān)視服務集成得很好，并提供了一個 REST API 來控制應用服務和查詢應用信息。具體表現(xiàn)如下：文章來源地址http://www.zghlxwxcb.cn/news/detail-634609.html

• Web UI 方式：Flink 提供了一個 Web UI 來觀察、監(jiān)視和調試正在運行的應用服務。并且還可以執(zhí)行或取消組件或任務的執(zhí)行。
• 日志集成服務：Flink 實現(xiàn)了流行的 slf4j 日志接口，并與日志框架 log4j 或 logback 集成。
• 指標服務：Flink 提供了一個復雜的度量系統(tǒng)來收集和報告系統(tǒng)和用戶定義的度量指標信息。度量信息可以導出到多個報表組件服務，包括 JMX，Ganglia，Graphite，Prometheus，StatsD，Datadog 和 Slf4j。
• 標準的 WEB REST API 接口服務：Flink 提供多種 REST API 接口，有提交新應用程序、獲取正在運行的應用程序的 Savepoint 服務信息、取消應用服務等接口。REST API 還提供元數(shù)據(jù)信息和已采集的運行中或完成后的應用服務的指標信息。

到了這里，關于【大數(shù)據(jù)】Flink 從入門到實踐（一）：初步介紹的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！