runners
Guides understanding and working with Apache Beam runners (Direct, Dataflow, Flink, Spark, etc.). Use when configuring pipelines for different execution environments or debugging runner-specific issues.
Datos de origen
- Repositorio
- apache/beam
- Última actividad en el origen
- 30 de abril de 2026 a las 20:13
- Idioma detectado de SKILL.md
- inglés
- Estrellas
- 8680
- Forks
- 4673
Opciones de instalación
De forma predeterminada está seleccionado el prompt que primero revisa el origen. Puedes cambiar a un comando directo o descargar una copia local.
Revisa los archivos de origen
Lee SKILL.md y los archivos complementarios que muestra SkillsMP antes de decidir si quieres instalarlo.
Mostrando SKILL.md
SKILL.md
Instrucciones de origen · Vista previa de solo lectura- name
- runners
- description
- Guides understanding and working with Apache Beam runners (Direct, Dataflow, Flink, Spark, etc.). Use when configuring pipelines for different execution environments or debugging runner-specific issues.
# 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 |
| Jet | `runners/jet/` | Hazelcast Jet |
| Twister2 | `runners/twister2/` | Twister2 |
## Direct Runner
For local development and testing.
### Java
```java
PipelineOptions options = PipelineOptionsFactory.create();
options.setRunner(DirectRunner.class);
Pipeline p = Pipeline.create(options);
```
### Python
```python
options = PipelineOptions()
options.view_as(StandardOptions).runner = 'DirectRunner'
p = beam.Pipeline(options=options)
```
### Command Line
```bash
--runner=DirectRunner
```
## Dataflow Runner
### Prerequisites
- GCP project with Dataflow API enabled
- Service account with Dataflow Admin role
- GCS bucket for staging
### Java Usage
```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 Usage
```python
options = PipelineOptions([
'--runner=DataflowRunner',
'--project=my-project',
'--region=us-central1',
'--temp_location=gs://my-bucket/temp'
])
```
### Runner v2
```bash
--experiments=use_runner_v2
```
### Custom SDK Container
```bash
--sdkContainerImage=gcr.io/project/beam_java11_sdk:custom
```
## Flink Runner
### Embedded Mode
```java
FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
options.setRunner(FlinkRunner.class);
options.setFlinkMaster("[local]");
```
### Cluster Mode
```java
options.setFlinkMaster("host:port");
```
### Portable Mode (Python)
```python
options = PipelineOptions([
'--runner=FlinkRunner',
'--flink_master=host:port',
'--environment_type=LOOPBACK' # or DOCKER, EXTERNAL
])
```
## Spark Runner
### Java
```java
SparkPipelineOptions options = PipelineOptionsFactory.as(SparkPipelineOptions.class);
options.setRunner(SparkRunner.class);
options.setSparkMaster("local[*]"); # or spark://host:port
```
### Python (Portable)
```python
options = PipelineOptions([
'--runner=SparkRunner',
'--spark_master_url=local[*]'
])
```
## Testing with Runners
### ValidatesRunner Tests
Tests that validate runner correctness:
```bash
# 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
```java
@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:
```bash
./gradlew :runners:flink:1.18:job-server:runShadow
```
Start Spark job server:
```bash
./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
```bash
./gradlew :runners:google-cloud-dataflow-java:worker:shadowJar
```
### Flink Job Server
```bash
./gradlew :runners:flink:1.18:job-server:shadowJar
```
### Spark Job Server
```bash
./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
Ver en GitHub