0
# Apache Airflow Providers Apache Beam
1
2
An Apache Airflow provider package that enables workflow orchestration and data processing capabilities by offering operators, hooks, and triggers for Apache Beam pipeline execution. Supports running Beam pipelines written in Python, Java, and Go across various runners including DirectRunner, DataflowRunner, SparkRunner, and FlinkRunner.
3
4
## Package Information
5
6
- **Package Name**: apache-airflow-providers-apache-beam
7
- **Package Type**: Python Package (PyPI)
8
- **Language**: Python
9
- **Installation**: `pip install apache-airflow-providers-apache-beam`
10
- **Version**: 6.1.3
11
- **Requires**: Apache Airflow >=2.10.0, Apache Beam >=2.60.0
12
13
## Core Imports
14
15
```python
16
from airflow.providers.apache.beam import __version__
17
```
18
19
Operators:
20
```python
21
from airflow.providers.apache.beam.operators.beam import (
22
BeamRunPythonPipelineOperator,
23
BeamRunJavaPipelineOperator,
24
BeamRunGoPipelineOperator,
25
BeamBasePipelineOperator
26
)
27
```
28
29
Hooks:
30
```python
31
from airflow.providers.apache.beam.hooks.beam import (
32
BeamHook,
33
BeamAsyncHook,
34
BeamRunnerType,
35
beam_options_to_args,
36
run_beam_command
37
)
38
```
39
40
Triggers:
41
```python
42
from airflow.providers.apache.beam.triggers.beam import (
43
BeamPythonPipelineTrigger,
44
BeamJavaPipelineTrigger,
45
BeamPipelineBaseTrigger
46
)
47
```
48
49
Version compatibility:
50
```python
51
from airflow.providers.apache.beam.version_compat import (
52
AIRFLOW_V_3_1_PLUS,
53
BaseHook,
54
BaseOperator
55
)
56
```
57
58
## Basic Usage
59
60
```python
61
from datetime import datetime, timedelta
62
from airflow import DAG
63
from airflow.providers.apache.beam.operators.beam import BeamRunPythonPipelineOperator
64
65
# Define default arguments
66
default_args = {
67
'owner': 'data-team',
68
'depends_on_past': False,
69
'start_date': datetime(2024, 1, 1),
70
'email_on_failure': False,
71
'email_on_retry': False,
72
'retries': 1,
73
'retry_delay': timedelta(minutes=5),
74
}
75
76
# Create DAG
77
dag = DAG(
78
'beam_pipeline_example',
79
default_args=default_args,
80
description='Run Apache Beam pipeline with Airflow',
81
schedule_interval=timedelta(days=1),
82
catchup=False,
83
)
84
85
# Define Beam pipeline task
86
run_beam_pipeline = BeamRunPythonPipelineOperator(
87
task_id='run_beam_pipeline',
88
py_file='gs://my-bucket/beam_pipeline.py',
89
runner='DataflowRunner',
90
pipeline_options={
91
'project': 'my-gcp-project',
92
'region': 'us-central1',
93
'temp_location': 'gs://my-bucket/temp',
94
'staging_location': 'gs://my-bucket/staging',
95
},
96
dataflow_config={
97
'job_name': 'my-beam-pipeline',
98
'project_id': 'my-gcp-project',
99
'location': 'us-central1',
100
},
101
dag=dag,
102
)
103
```
104
105
## Architecture
106
107
The provider follows Airflow's standard architecture pattern with three main component types:
108
109
- **Operators**: Execute Beam pipelines as Airflow tasks, supporting Python, Java, and Go implementations
110
- **Hooks**: Provide low-level interface to Apache Beam, handling pipeline execution and monitoring
111
- **Triggers**: Enable deferrable execution for long-running pipelines with asynchronous monitoring
112
113
The provider integrates with Google Cloud Dataflow when the google provider is available, enabling cloud-scale pipeline execution with monitoring and job management capabilities.
114
115
## Capabilities
116
117
### Python Pipeline Execution
118
119
Execute Apache Beam pipelines written in Python with support for virtual environments, custom requirements, and various runners including local DirectRunner and cloud DataflowRunner.
120
121
```python { .api }
122
class BeamRunPythonPipelineOperator(BeamBasePipelineOperator):
123
def __init__(
124
self,
125
*,
126
py_file: str,
127
runner: str = "DirectRunner",
128
default_pipeline_options: dict | None = None,
129
pipeline_options: dict | None = None,
130
py_interpreter: str = "python3",
131
py_options: list[str] | None = None,
132
py_requirements: list[str] | None = None,
133
py_system_site_packages: bool = False,
134
gcp_conn_id: str = "google_cloud_default",
135
dataflow_config: DataflowConfiguration | dict | None = None,
136
deferrable: bool = False,
137
**kwargs,
138
) -> None: ...
139
```
140
141
[Python Pipelines](./python-pipelines.md)
142
143
### Java Pipeline Execution
144
145
Execute Apache Beam pipelines written in Java using self-executing JAR files with support for various runners and Dataflow integration.
146
147
```python { .api }
148
class BeamRunJavaPipelineOperator(BeamBasePipelineOperator):
149
def __init__(
150
self,
151
*,
152
jar: str,
153
runner: str = "DirectRunner",
154
job_class: str | None = None,
155
default_pipeline_options: dict | None = None,
156
pipeline_options: dict | None = None,
157
gcp_conn_id: str = "google_cloud_default",
158
dataflow_config: DataflowConfiguration | dict | None = None,
159
deferrable: bool = False,
160
**kwargs,
161
) -> None: ...
162
```
163
164
[Java Pipelines](./java-pipelines.md)
165
166
### Go Pipeline Execution
167
168
Execute Apache Beam pipelines written in Go from source files or pre-compiled binaries with support for cross-platform execution.
169
170
```python { .api }
171
class BeamRunGoPipelineOperator(BeamBasePipelineOperator):
172
def __init__(
173
self,
174
*,
175
go_file: str = "",
176
launcher_binary: str = "",
177
worker_binary: str = "",
178
runner: str = "DirectRunner",
179
default_pipeline_options: dict | None = None,
180
pipeline_options: dict | None = None,
181
gcp_conn_id: str = "google_cloud_default",
182
dataflow_config: DataflowConfiguration | dict | None = None,
183
**kwargs,
184
) -> None: ...
185
```
186
187
[Go Pipelines](./go-pipelines.md)
188
189
### Pipeline Monitoring and Hooks
190
191
Low-level interface for executing and monitoring Apache Beam pipelines with both synchronous and asynchronous execution modes.
192
193
```python { .api }
194
class BeamHook(BaseHook):
195
def __init__(self, runner: str) -> None: ...
196
def start_python_pipeline(
197
self,
198
variables: dict,
199
py_file: str,
200
py_options: list[str],
201
py_interpreter: str = "python3",
202
py_requirements: list[str] | None = None,
203
py_system_site_packages: bool = False,
204
process_line_callback: Callable[[str], None] | None = None,
205
is_dataflow_job_id_exist_callback: Callable[[], bool] | None = None,
206
) -> None: ...
207
def start_java_pipeline(
208
self,
209
variables: dict,
210
jar: str,
211
job_class: str | None = None,
212
process_line_callback: Callable[[str], None] | None = None,
213
is_dataflow_job_id_exist_callback: Callable[[], bool] | None = None,
214
) -> None: ...
215
def start_go_pipeline(
216
self,
217
variables: dict,
218
go_file: str,
219
process_line_callback: Callable[[str], None] | None = None,
220
should_init_module: bool = False,
221
) -> None: ...
222
223
class BeamAsyncHook(BeamHook):
224
def __init__(self, runner: str) -> None: ...
225
async def start_python_pipeline_async(
226
self,
227
variables: dict,
228
py_file: str,
229
py_options: list[str] | None = None,
230
py_interpreter: str = "python3",
231
py_requirements: list[str] | None = None,
232
py_system_site_packages: bool = False,
233
process_line_callback: Callable[[str], None] | None = None,
234
) -> int: ...
235
async def start_java_pipeline_async(
236
self,
237
variables: dict,
238
jar: str,
239
job_class: str | None = None,
240
process_line_callback: Callable[[str], None] | None = None,
241
) -> int: ...
242
```
243
244
[Hooks and Monitoring](./hooks-monitoring.md)
245
246
### Asynchronous Pipeline Triggers
247
248
Deferrable execution triggers for long-running pipelines that enable efficient resource utilization by yielding control during pipeline execution.
249
250
```python { .api }
251
class BeamPythonPipelineTrigger(BeamPipelineBaseTrigger):
252
def __init__(
253
self,
254
variables: dict,
255
py_file: str,
256
py_options: list[str] | None = None,
257
py_interpreter: str = "python3",
258
py_requirements: list[str] | None = None,
259
py_system_site_packages: bool = False,
260
runner: str = "DirectRunner",
261
gcp_conn_id: str = "google_cloud_default",
262
) -> None: ...
263
def serialize(self) -> tuple[str, dict[str, Any]]: ...
264
async def run(self) -> AsyncIterator[TriggerEvent]: ...
265
266
class BeamJavaPipelineTrigger(BeamPipelineBaseTrigger):
267
def __init__(
268
self,
269
variables: dict,
270
jar: str,
271
job_class: str | None = None,
272
runner: str = "DirectRunner",
273
gcp_conn_id: str = "google_cloud_default",
274
) -> None: ...
275
def serialize(self) -> tuple[str, dict[str, Any]]: ...
276
async def run(self) -> AsyncIterator[TriggerEvent]: ...
277
```
278
279
[Triggers and Deferrable Execution](./triggers.md)
280
281
### Version Compatibility
282
283
Cross-version compatibility components that provide stable imports and version detection for different Airflow releases.
284
285
```python { .api }
286
AIRFLOW_V_3_1_PLUS: bool
287
"""Boolean flag indicating Airflow version 3.1+ compatibility."""
288
289
# Version-compatible base classes (imported from appropriate Airflow version)
290
BaseHook: type
291
"""Version-compatible BaseHook class."""
292
293
BaseOperator: type
294
"""Version-compatible BaseOperator class."""
295
```
296
297
## Types
298
299
```python { .api }
300
class BeamBasePipelineOperator(BaseOperator):
301
"""
302
Abstract base class for all Apache Beam pipeline operators.
303
304
Provides common functionality including pipeline option handling,
305
Dataflow integration, and error management.
306
"""
307
template_fields = ["runner", "pipeline_options", "default_pipeline_options", "dataflow_config"]
308
309
def __init__(
310
self,
311
*,
312
runner: str = "DirectRunner",
313
default_pipeline_options: dict | None = None,
314
pipeline_options: dict | None = None,
315
gcp_conn_id: str = "google_cloud_default",
316
dataflow_config: DataflowConfiguration | dict | None = None,
317
deferrable: bool = False,
318
**kwargs,
319
) -> None: ...
320
321
def execute(self, context: Context) -> dict: ...
322
def execute_on_dataflow(self, context: Context) -> dict: ...
323
def on_kill(self) -> None: ...
324
325
class BeamRunnerType:
326
"""Helper class for listing available runner types."""
327
DataflowRunner = "DataflowRunner"
328
DirectRunner = "DirectRunner"
329
SparkRunner = "SparkRunner"
330
FlinkRunner = "FlinkRunner"
331
SamzaRunner = "SamzaRunner"
332
NemoRunner = "NemoRunner"
333
JetRunner = "JetRunner"
334
Twister2Runner = "Twister2Runner"
335
336
class DataflowConfiguration:
337
"""
338
Configuration object for Dataflow-specific options.
339
340
Used to configure Google Cloud Dataflow execution parameters
341
including job naming, project settings, and execution behavior.
342
"""
343
job_name: str
344
project_id: str
345
location: str
346
gcp_conn_id: str = "google_cloud_default"
347
wait_until_finished: bool = True
348
poll_sleep: int = 10
349
cancel_timeout: int = 300
350
drain_pipeline: bool = False
351
service_account: str | None = None
352
impersonation_chain: list[str] | None = None
353
check_if_running: str = "WaitForRun"
354
multiple_jobs: bool = False
355
356
class Context:
357
"""Airflow execution context for task instances."""
358
359
def beam_options_to_args(options: dict) -> list[str]:
360
"""
361
Convert pipeline options dictionary to command line arguments.
362
363
Args:
364
options: Dictionary with pipeline options
365
366
Returns:
367
List of formatted command line arguments
368
"""
369
370
def run_beam_command(
371
cmd: list[str],
372
log: logging.Logger,
373
process_line_callback: Callable[[str], None] | None = None,
374
working_directory: str | None = None,
375
is_dataflow_job_id_exist_callback: Callable[[], bool] | None = None,
376
) -> None:
377
"""
378
Execute Beam pipeline command in subprocess with monitoring.
379
380
Args:
381
cmd: Command parts to execute
382
log: Logger for output
383
process_line_callback: Optional output processor
384
working_directory: Execution directory
385
is_dataflow_job_id_exist_callback: Optional job ID detector
386
"""
387
388
# Version compatibility support
389
AIRFLOW_V_3_1_PLUS: bool
390
"""Boolean flag indicating Airflow version 3.1+ compatibility."""
391
392
# Trigger types
393
class TriggerEvent:
394
"""Event yielded by triggers to indicate status changes."""
395
def __init__(self, payload: dict[str, Any]) -> None: ...
396
397
class AsyncIterator[T]:
398
"""Async iterator type for trigger events."""
399
400
class NamedTemporaryFile:
401
"""Temporary file with a visible name in the file system."""
402
name: str
403
404
# Type aliases
405
Any = typing.Any
406
Callable = typing.Callable
407
```