runners

用途

Apache Beam Runners 是 Apache Beam 中负责在分布式处理后端上执行管道的组件。每个 Runner 会把可移植的 Beam 模型翻译成其原生执行引擎的作业。本技能用于指导理解并配置不同后端上的 Beam 管道。

可用 Runner

Runner	位置	说明
Direct	runners/direct-java/	本地执行，用于测试
Prism	runners/prism/	可移植的本地 Runner
Dataflow	runners/google-cloud-dataflow-java/	Google Cloud Dataflow
Flink	runners/flink/	Apache Flink
Spark	runners/spark/	Apache Spark
Jet	runners/jet/	Hazelcast Jet
Twister2	runners/twister2/	Twister2

Direct Runner

用于本地开发与测试。

Java：

PipelineOptions options = PipelineOptionsFactory.create();
options.setRunner(DirectRunner.class);
Pipeline p = Pipeline.create(options);

Python：

options = PipelineOptions()
options.view_as(StandardOptions).runner = 'DirectRunner'
p = beam.Pipeline(options=options)

命令行：

--runner=DirectRunner

Dataflow Runner

前置条件：

已启用 Dataflow API 的 GCP 项目
具有 Dataflow Admin 角色的服务账号
用于暂存的 GCS 存储桶

Java 用法：

DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
options.setRunner(DataflowRunner.class);
options.setProject("my-project");
options.setRegion("us-central1");
options.setTempLocation("gs://my-bucket/temp");

Python 用法：

options = PipelineOptions([
    '--runner=DataflowRunner',
    '--project=my-project',
    '--region=us-central1',
    '--temp_location=gs://my-bucket/temp'
])

Runner v2：

--experiments=use_runner_v2

自定义 SDK 容器：

--sdkContainerImage=gcr.io/project/beam_java11_sdk:custom

Flink Runner

嵌入式模式：

FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
options.setRunner(FlinkRunner.class);
options.setFlinkMaster("[local]");

集群模式：

options.setFlinkMaster("host:port");

可移植模式（Python）：

options = PipelineOptions([
    '--runner=FlinkRunner',
    '--flink_master=host:port',
    '--environment_type=LOOPBACK'  # or DOCKER, EXTERNAL
])

Spark Runner

Java：

SparkPipelineOptions options = PipelineOptionsFactory.as(SparkPipelineOptions.class);
options.setRunner(SparkRunner.class);
options.setSparkMaster("local[*]");  # or spark://host:port

Python（可移植）：

options = PipelineOptions([
    '--runner=SparkRunner',
    '--spark_master_url=local[*]'
])

使用 Runner 进行测试

ValidatesRunner 测试用于验证 Runner 正确性：

# Direct Runner
./gradlew :runners:direct-java:validatesRunner

# Flink Runner
./gradlew :runners:flink:1.18:validatesRunner

# Spark Runner
./gradlew :runners:spark:3:validatesRunner

# Dataflow Runner
./gradlew :runners:google-cloud-dataflow-java:validatesRunner

配合 Runner 使用 TestPipeline：

@Rule public TestPipeline pipeline = TestPipeline.create();

// Set runner via system property
-DbeamTestPipelineOptions='["--runner=TestDataflowRunner"]'

可移植 Runner

概念：

通过 Fn API 实现与 SDK 无关的执行
SDK 在容器中运行，通过 gRPC 通信

环境类型：

DOCKER — SDK 位于 Docker 容器
LOOPBACK — SDK 位于同一进程（测试）
EXTERNAL — SDK 位于指定地址
PROCESS — SDK 位于子进程

Job Server：

启动 Flink job server：

./gradlew :runners:flink:1.18:job-server:runShadow

启动 Spark job server：

./gradlew :runners:spark:3:job-server:runShadow

Runner 专属选项

Dataflow：

选项	说明
--project	GCP 项目
--region	GCP 区域
--tempLocation	GCS 临时位置
--stagingLocation	GCS 暂存位置
--numWorkers	初始 worker 数
--maxNumWorkers	最大 worker 数
--workerMachineType	VM 类型

Flink：

选项	说明
--flinkMaster	Flink master 地址
--parallelism	默认并行度
--checkpointingInterval	检查点间隔

Spark：

选项	说明
--sparkMaster	Spark master URL
--sparkConf	额外的 Spark 配置

构建 Runner 产物

Dataflow Worker Jar：

./gradlew :runners:google-cloud-dataflow-java:worker:shadowJar

Flink Job Server：

./gradlew :runners:flink:1.18:job-server:shadowJar

Spark Job Server：

./gradlew :runners:spark:3:job-server:shadowJar

调试

Direct Runner：

启用日志：-Dorg.slf4j.simpleLogger.defaultLogLevel=debug
使用 --targetParallelism=1 获得确定性执行