Apache Beam Runners

Overview

Runners execute Beam pipelines on distributed processing backends. Each runner translates the portable Beam model to its native execution engine.

Available Runners

Runner Location Description

Direct runners/direct-java/

Local execution for testing

Prism runners/prism/

Portable local runner

Dataflow runners/google-cloud-dataflow-java/

Google Cloud Dataflow

Flink runners/flink/

Apache Flink

Spark runners/spark/

Apache Spark

Samza runners/samza/

Apache Samza

Jet runners/jet/

Hazelcast Jet

Twister2 runners/twister2/

Twister2

Direct Runner

For local development and testing.

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)

Command Line

--runner=DirectRunner

Dataflow Runner

Prerequisites

• GCP project with Dataflow API enabled

• Service account with Dataflow Admin role

• GCS bucket for staging

Java Usage

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 Usage

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

Runner v2

--experiments=use_runner_v2

Custom SDK Container

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

Flink Runner

Embedded Mode

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

Cluster Mode

options.setFlinkMaster("host:port");

Portable Mode (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 (Portable)

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

Testing with Runners

ValidatesRunner Tests

Tests that validate runner correctness:

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

TestPipeline with Runners

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

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

Portable Runners

Concept

• SDK-independent execution via Fn API

• SDK runs in container, communicates via gRPC

Environment Types

• DOCKER

• SDK in Docker container

• LOOPBACK

• SDK in same process (testing)

• EXTERNAL

• SDK at specified address

• PROCESS

• SDK in subprocess

Job Server

Start Flink job server:

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

Start Spark job server:

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

Runner-Specific Options

Dataflow

Option Description

--project

GCP project

--region

GCP region

--tempLocation

GCS temp location

--stagingLocation

GCS staging

--numWorkers

Initial workers

--maxNumWorkers

Max workers

--workerMachineType

VM type

Flink

Option Description

--flinkMaster

Flink master address

--parallelism

Default parallelism

--checkpointingInterval

Checkpoint interval

Spark

Option Description

--sparkMaster

Spark master URL

--sparkConf

Additional Spark config

Building Runner Artifacts

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

Debugging

Direct Runner

• Enable logging: -Dorg.slf4j.simpleLogger.defaultLogLevel=debug

• Use --targetParallelism=1 for deterministic execution

Dataflow

• Check Dataflow UI: console.cloud.google.com/dataflow

• Use --experiments=upload_graph for graph debugging

• Worker logs in Cloud Logging

Portable Runners

• Enable debug logging on job server

• Check SDK harness logs in worker containers