概述

Apache Flink 執行器可用於使用 Apache Flink 執行 Beam 管線。對於執行，您可以選擇叢集執行模式（例如 Yarn/Kubernetes/Mesos）或本機嵌入式執行模式，這對於測試管線很有用。

Flink 執行器和 Flink 適用於大規模、連續的作業，並提供

• 一個以串流為先的執行時環境，同時支援批次處理和資料串流程式
• 一個同時支援極高輸送量和低事件延遲的執行時環境
• 具有精確一次處理保證的容錯能力
• 串流程式中的自然回壓
• 自訂記憶體管理，可在記憶體內和核心外資料處理演算法之間進行高效且穩健的切換
• 與 YARN 和 Apache Hadoop 生態系統的其他元件整合

使用 Apache Flink 執行器

重要的是要了解 Flink 執行器有兩種版本

1. 原始的經典執行器，僅支援 Java（和其他基於 JVM 的語言）
2. 較新的可攜式執行器，支援 Java/Python/Go

您可能會問為什麼會有兩個執行器？

Beam 及其執行器最初僅支援基於 JVM 的語言（例如 Java/Scala/Kotlin）。Python 和 Go SDK 是稍後才添加的。必須顯著更改執行器的架構，才能支援執行以其他語言編寫的管線。

如果您的應用程式僅使用 Java，那麼您目前應該使用經典執行器。最終，可攜式執行器將取代經典執行器，因為它包含用於在未來執行 Java、Python、Go 和更多語言的通用框架。

如果您想在 Flink 上使用 Beam 執行 Python 管線，則需要使用可攜式執行器。有關可攜性的更多資訊，請瀏覽可攜性頁面。

因此，本指南分為多個部分，以記錄 Flink 執行器的經典和可攜式功能。此外，Python 提供方便的封裝器來處理執行器的完整生命週期，因此進一步分為自動（建議）或手動管理可攜性元件。請使用下面的切換器為執行器選擇適當的模式

先決條件和設定

如果您想使用 Flink 執行器的本機執行模式，則不必完成任何叢集設定。您可以直接執行您的 Beam 管線。請務必將執行器設定為FlinkRunnerPortableRunner。

若要使用 Flink 執行器在叢集上執行，您必須按照 Flink 設定快速入門來設定 Flink 叢集。

相依性

您必須在 pom.xml 或 build.gradle 中指定您對 Flink 執行器的依賴性。使用下面相容性表格中的 Beam 版本和構件 ID。例如

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-flink-1.18</artifactId>
  <version>2.60.0</version>
</dependency>

您的執行環境中需要安裝 Docker。若要執行嵌入式 Flink 叢集或將 Flink 執行器用於 Python < 3.6，您的執行環境中還需要有 Java。

您的執行環境中需要安裝 Docker。

在 Flink 叢集上執行 Beam 管線

若要在 Flink 叢集上執行管線，您需要將您的程式與所有相依性一起封裝在所謂的 fat jar 中。如何執行此操作取決於您的建置系統，但是如果您按照Beam 快速入門進行操作，則這是您必須執行的命令

$ mvn package -Pflink-runner

在此命令的 target 資料夾中尋找輸出 JAR。

Beam 快速入門 Maven 專案設定為使用 Maven Shade 外掛程式來建立 fat jar，而 -Pflink-runner 引數確保包含對 Flink 執行器的依賴性。

若要執行管線，最簡單的選項是使用 Flink 的一部分 flink 命令

$ bin/flink run -c org.apache.beam.examples.WordCount /path/to/your.jar –runner=FlinkRunner –other-parameters

或者，您也可以使用 Maven 的 exec 命令。例如，若要執行 WordCount 範例

mvn exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
    -Pflink-runner \
    -Dexec.args="--runner=FlinkRunner \
      --inputFile=/path/to/pom.xml \
      --output=/path/to/counts \
      --flinkMaster=<flink master url> \
      --filesToStage=target/word-count-beam-bundled-0.1.jar"

如果您的本機電腦上正在執行 Flink JobManager，您可以為 flinkMaster 提供 localhost:8081。否則，將為該作業啟動一個嵌入式 Flink 叢集。

若要在 Flink 上執行管線，請將執行器設定為 FlinkRunner，並將 flink_master 設定為 Flink 叢集的 master URL。此外，還可以選擇將 environment_type 設定為 LOOPBACK。例如，在啟動本機 Flink 叢集後，可以執行

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions([
    "--runner=FlinkRunner",
    "--flink_master=localhost:8081",
    "--environment_type=LOOPBACK"
])
with beam.Pipeline(options) as p:
    ...

若要在嵌入式 Flink 叢集上執行，只需省略 flink_master 選項，系統將為該作業自動啟動和關閉嵌入式 Flink 叢集。

對於舊版 Python，也可能需要選擇性的 flink_version 選項。

從 Beam 2.18.0 開始，預先建置的 Flink Job Service Docker 映像可在 Docker Hub 上取得：Flink 1.16。Flink 1.17。Flink 1.18。

若要在嵌入式 Flink 叢集上執行管線

（1）啟動 JobService 端點：docker run --net=host apache/beam_flink1.10_job_server:latest

JobService 是您將 Beam 管線提交到的中央執行個體。JobService 將為管線建立 Flink 作業，並執行該作業。

（2）使用 PortableRunner 將 Python 管線提交至上述端點，job_endpoint 設定為 localhost:8099（這是 JobService 的預設位址）。還可以選擇將 environment_type 設定為 LOOPBACK。例如

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions([
    "--runner=PortableRunner",
    "--job_endpoint=localhost:8099",
    "--environment_type=LOOPBACK"
])
with beam.Pipeline(options) as p:
    ...

若要在單獨的Flink 叢集上執行

（1）啟動一個公開 Rest 介面的 Flink 叢集（例如，預設為 localhost:8081）。

（2）使用 Flink Rest 端點啟動 JobService：docker run --net=host apache/beam_flink1.10_job_server:latest --flink-master=localhost:8081。

（3）如上所述提交管線。

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions([
    "--runner=PortableRunner",
    "--job_endpoint=localhost:8099",
    "--environment_type=LOOPBACK"
])
with beam.Pipeline(options=options) as p:
    ...

請注意，environment_type=LOOPBACK 僅適用於本機測試，不適用於遠端叢集。有關詳細資訊，請參閱此處。

其他資訊和注意事項

監控您的作業

您可以使用 Flink JobManager 儀表板或其 Rest 介面來監控正在執行的 Flink 作業。預設情況下，這可在 JobManager 節點的連接埠 8081 上取得。如果您在本機電腦上安裝了 Flink，則該位址會是 https://127.0.0.1:8081。注意：當您使用 [local] 模式時，將會啟動一個嵌入式 Flink 叢集，不會提供儀表板。

串流執行

如果您的管線使用無界資料來源或接收器，則 Flink 執行器會自動切換到串流模式。您可以使用 --streaming 旗標強制執行串流模式。

注意：當使用無界來源且未啟用檢查點時，執行器會列印警告訊息。許多來源（例如 PubSubIO）依賴其檢查點來進行確認，這僅能在為 FlinkRunner 啟用檢查點時完成。若要啟用檢查點，請將checkpointingIntervalcheckpointing_interval設定為所需的檢查點間隔（以毫秒為單位）。

Flink 執行器的管線選項

當您使用 Flink 執行器執行管線時，您可以設定這些管線選項。

以下 Flink 特定的管線選項清單是從 FlinkPipelineOptions 參考類別自動產生

有關一般的 Beam Pipeline 選項，請參閱 PipelineOptions 參考資料。

Flink 版本相容性

Flink 叢集版本必須與 FlinkRunner 使用的次要版本相符。次要版本是版本字串中的前兩個數字，例如在 1.18.0 中，次要版本是 1.18。

我們會嘗試追蹤 Beam 發佈時 Apache Flink 的最新版本。Beam 支援 Flink 版本的時間與 Flink 社群支援的時間相同。Flink 社群支援最新的兩個次要版本。當停止支援 Flink 版本時，也可能會從 Beam 中移除該版本並將其標示為已淘汰。若要找出哪個版本的 Flink 與 Beam 相容，請參閱下表

若要取得正確的 Flink 版本，請參閱 Flink 下載頁面。

如需更多資訊，Flink 文件可能會很有幫助。

Beam 功能

Beam 功能矩陣記錄了傳統 Flink Runner 的功能。

可攜式功能矩陣記錄了可攜式 Flink Runner 的功能。