概述

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 集群或对 Python < 3.6 使用 Flink 运行器，您还需要在您的执行环境中提供 java。

您需要在您的执行环境中安装 Docker。

在 Flink 集群上执行 Beam 管道

要在 Flink 集群上执行管道，您需要将您的程序与所有依赖项打包到所谓的胖 JAR 中。如何执行此操作取决于您的构建系统，但是如果您按照 Beam 快速入门 进行操作，则需要运行以下命令

$ mvn package -Pflink-runner

在 target 文件夹中查找此命令的输出 JAR。

Beam 快速入门 Maven 项目设置为使用 Maven Shade 插件来创建胖 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 集群的主机 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://:8081。注意：当您使用 [local] 模式时，将启动一个嵌入式 Flink 集群，它不会提供仪表板。

流式执行

如果您的管道使用无界数据源或接收器，Flink 运行器将自动切换到流式模式。您可以使用 --streaming 标志强制执行流式模式。

注意：当使用无界源且未启用检查点时，运行器将打印警告消息。许多源（如 PubSubIO）依赖于它们的检查点以进行确认，这只能在为 FlinkRunner 启用检查点时完成。要启用检查点，请将 checkpointingIntervalcheckpointing_interval 设置为所需的检查点间隔（以毫秒为单位）。

Flink 运行器的管道选项

使用 Flink 运行器执行管道时，您可以设置这些管道选项。

以下 Flink 特定管道选项列表是根据 FlinkPipelineOptions 引用类自动生成的

有关一般 Beam 管道选项，请参阅 PipelineOptions 参考。

Flink 版本兼容性

Flink 集群版本必须与 FlinkRunner 使用的次版本匹配。次版本是版本字符串中的前两个数字，例如在 1.18.0 中，次版本是 1.18。

我们尝试跟踪 Beam 发布时 Apache Flink 的最新版本。Flink 版本在 Flink 社区支持的时间内受 Beam 支持。Flink 社区支持最近两个次版本。当 Flink 版本的支持被取消时，它可能会被弃用并从 Beam 中删除。要了解 Beam 与哪个版本的 Flink 兼容，请参阅下表。

要获取正确的 Flink 版本，请参阅 Flink 下载页面。

有关更多信息，Flink 文档 可能会有帮助。

Beam 功能

Beam 功能矩阵 记录了经典 Flink Runner 的功能。

可移植功能矩阵 记录了可移植 Flink Runner 的功能。