Apache Beam SDK for Python

Warnings

• breaking As of Apache Beam 2.70.0, many Python dependencies were split into 'extras'. If your pipeline previously relied on these implicitly, you may need to explicitly install them using `pip install apache-beam[gcp,interactive,yaml,redis,hadoop,tfrecord,...]` to ensure all necessary components are present.
  Fix: Review your pipeline's dependencies and install Apache Beam with the appropriate extras, e.g., `pip install 'apache-beam[gcp,interactive]'`.
• deprecated Support for Python 3.9 was removed in Apache Beam 2.70.0 after Python 3.9 reached its End-of-Life in October 2025. Ensure your development and runtime environments use Python 3.10 or newer.
  Fix: Upgrade your Python environment to 3.10 or a later supported version (e.g., Python 3.10, 3.11, 3.12). The PyPI package requires >=3.10 for 2.71.0.
• gotcha The `dill` library is no longer a required, default dependency as of Apache Beam 2.71.0. If your pipeline explicitly uses the `pickle_library=dill` pipeline option, you must manually ensure `dill==0.3.1.1` is installed in both your submission and runtime environments.
  Fix: If using `pickle_library=dill`, add `dill==0.3.1.1` to your `requirements.txt` or ensure it's installed in your custom container.
• gotcha Apache Beam's distributed nature makes global state problematic. Avoid using global variables to share data across elements. Instead, leverage Beam's side inputs for read-only shared data or stateful processing primitives (`StateSpec`, `TimerSpec`) for mutable, per-key state.
  Fix: Refactor code to use Beam's recommended patterns for sharing data, such as side inputs or stateful `DoFn`s, which are designed for distributed execution.
• gotcha Managing Python pipeline dependencies can lead to 'Diamond Dependency' problems, especially when bundling `apache-beam` with other libraries or using many optional extras. Incompatible versions of shared dependencies can cause runtime errors.
  Fix: Define pipeline dependencies carefully using `requirements.txt` (or `setup.py` for packages). Consider using custom container images to control the exact environment and pre-install dependencies, ensuring reproducibility and avoiding conflicts.

Install

• pip install apache-beam Basic installation
• pip install 'apache-beam[gcp]' With Google Cloud Platform (Dataflow) support

Imports

• beam

  import apache_beam as beam

  Standard alias for the Apache Beam SDK.
• ReadFromText

  from apache_beam.io import ReadFromText

• WriteToText

  from apache_beam.io import WriteToText

• PipelineOptions

  from apache_beam.options.pipeline_options import PipelineOptions

Quickstart

This classic WordCount example demonstrates basic Apache Beam concepts: reading data from a source (local file or GCS), applying transformations like splitting, mapping, and combining, and writing the results to an output file. Run it locally using the DirectRunner.

import re
import argparse
import apache_beam as beam
from apache_beam.io import ReadFromText, WriteToText
from apache_beam.options.pipeline_options import PipelineOptions

def main(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--input',
        dest='input',
        default='gs://dataflow-samples/shakespeare/kinglear.txt',
        help='Input file to process.')
    parser.add_argument(
        '--output',
        dest='output',
        default='output.txt',
        help='Output file to write results to.')
    known_args, pipeline_args = parser.parse_known_args(argv)

    pipeline_options = PipelineOptions(pipeline_args)

    with beam.Pipeline(options=pipeline_options) as p:
        # Read the text file into a PCollection
        lines = p | ReadFromText(known_args.input)

        # Count the occurrences of each word
        counts = (
            lines
            | 'Split' >> beam.FlatMap(lambda x: re.findall(r'[A-Za-z\']+', x))
            | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
            | 'GroupAndSum' >> beam.CombinePerKey(sum)
        )

        # Format the counts into strings
        output = counts | 'Format' >> beam.Map(
            lambda word_count: '%s: %s' % (word_count[0], word_count[1]))

        # Write the output
        output | WriteToText(known_args.output)

if __name__ == '__main__':
    print("Running Beam WordCount pipeline locally...")
    main()
    print("Pipeline finished. Check 'output.txt' for results.")