Apache Beam Architecture

1. Introduction

2. Why This Concept Exists

Before Apache Beam's modern architecture, developers faced a major limitation: runners were engine-specific. If you wrote a Spark job, it ran only on a JVM environment because Spark was built in Scala/Java. Running Python code required complex wrapper processes, which degraded performance.

3. Key Terminology

• Runner API: A language-independent definition of a pipeline graph, serialized using Protocol Buffers.
• Fn API: The gRPC interface that allows runners (written in Java/Scala) to communicate with user-defined code running in language-specific containers (Workers).
• Worker Harness: A lightweight execution container spun up by the runner to run your Python/Go code during pipeline execution.
• Logical vs. Physical DAG: Logical DAG is the language-agnostic representation built by the SDK. Physical DAG is the execution graph optimized for Flink, Spark, or Dataflow.

4. How It Works

1. Logical Compilation: The Python SDK runs your code to construct a pipeline graph, checking syntax and types. It outputs a serialized Protocol Buffer representation (Runner API proto).
2. Job Submission: The proto is sent to the runner's Job Service.
3. Physical Compilation: The runner translates the proto into a physical execution graph optimized for its environment (e.g., Flink Task Managers).
4. Fn API Execution: When executing steps, the runner launches a containerized Worker Harness (via Docker). The runner streams data elements over gRPC to the worker harness, which executes your custom Python functions (like DoFn or lambdas) and returns the output elements.

5. Visual Diagram

6. Code Example

pythonCopy
import apache_beam as beam

class ConfiguredFilterDoFn(beam.DoFn):
    def __init__(self, prefix):
        # Definition-time constructor: Runs locally on the client machine
        self.prefix = prefix
        self.counter = None

    def setup(self):
        # Execution-time setup: Runs on the remote worker machine inside the harness
        self.counter = 0

    def process(self, element):
        # Execution-time processing: Called for every element in the harness
        if element.startswith(self.prefix):
            self.counter += 1
            yield f"MATCH #{self.counter}: {element}"

# Run locally using DirectRunner
with beam.Pipeline() as p:
    (p 
     | "CreateLogs" >> beam.Create(["ERR_auth_failed", "INFO_connected", "ERR_db_timeout"])
     | "FilterErrors" >> beam.ParDo(ConfiguredFilterDoFn("ERR"))
     | "Print" >> beam.Map(print))

7. Code Explanation

• __init__(self, prefix) is executed during pipeline definition time on the local developer machine. The prefix variable is serialized and bundled into the Pipeline Proto.
• setup(self) is executed on the remote worker node once the container starts up. This is where you would instantiate database connections or load models.
• process(self, element) is executed for every element sent over the Fn API gRPC socket from the runner process to the worker harness.

8. Real Production Example

When running a Python Beam pipeline on a Google Cloud Dataflow cluster, the Dataflow service acts as the Java runner. Dataflow provisions virtual machines that run a Java runner agent. This agent spins up a Docker container containing the Beam Python SDK worker harness. The Java runner reads data from sources, streams elements via gRPC to the Python container, and collects the processed results to write to BigQuery.

9. Common Mistakes

• Instantiating Heavy Objects in __init__: Initializing non-serializable objects (like socket connections, databases, or large ML models) inside the __init__ constructor. This causes serialization errors (like PicklingError) during logical DAG packaging. Heavy objects must always be initialized in the setup() method.
• Assuming Worker Local Persistence: Writing to local files inside a DoFn and assuming they will persist. Workers are ephemeral and run in isolated Docker containers.

10. Best Practices

• Use setup() or start_bundle() for one-time initialization task blocks (like opening database sessions).
• Avoid using global state variables in Python modules as they will not be shared across multi-threaded or distributed workers.

11. Summary

• Beam splits compilation (SDK) from execution (Runner).
• The Runner API serializes the logical graph using Protocol Buffers.
• The Fn API standardizes runtime communication via gRPC.
• User code executes inside isolated worker harness containers.

12. Interactive Challenges

pythonCopy
class DBReadDoFn(beam.DoFn):
    def __init__(self):
        pass
    # TODO: Add the setup method to initialize self.db

13. Related Content

• Pipeline Runners
• Direct Runner
• Beam Programming Model