Skip to content

Apache Beam — Unified Batch & Stream Processing with Portable Pipelines

DodaTech Updated 2026-06-23 11 min read

In this tutorial, you'll learn about Apache Beam. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Apache Beam is a unified programming model for batch and Stream Processing that lets you write a pipeline once and run it on any execution engine — Flink, Spark, Dataflow, or direct runner — using portable, language-agnostic pipeline definitions.

What You'll Learn

By the end of this tutorial, you'll understand Beam's core abstractions (Pipeline, PCollection, PTransform), how to build batch and streaming pipelines, windowing and triggers for out-of-order events, I/O connectors for reading and writing data, and how to run pipelines on different runners.

Why It Matters

Most organizations maintain separate codebases for batch jobs (Spark, dbt, Airflow) and streaming jobs (Flink, Kafka Streams). Beam eliminates this duplication with a single API that works for both. You write the pipeline once — Beam handles the execution semantics based on whether you're processing bounded or unbounded data. DodaTech uses Beam for cross-platform data processing workloads, running the same pipeline on Dataflow for cloud and Flink for on-premise deployments.

Real-World Use

Google Cloud Dataflow (Beam-based) processes 100+ PB/day for Google's internal analytics. Netflix uses Beam for its data processing pipeline. Zalando runs Beam pipelines for real-time inventory management across Europe.

Beam Pipeline Architecture

flowchart TB
    subgraph "Pipeline Construction"
        A[Pipeline.create] --> B[Apply Transforms]
        B --> C[PCollection]
        C --> D[ParDo / GroupByKey]
    end
    subgraph "Runner Selection"
        E[DirectRunner] --> F[Local Test]
        G[FlinkRunner] --> H[Streaming Cluster]
        I[DataflowRunner] --> J[Cloud Managed]
        K[SparkRunner] --> L[Spark Cluster]
    end
    subgraph "Execution"
        F --> M[Output]
        H --> M
        J --> M
        L --> M
    end
    style A fill:#f90,color:#fff
    style C fill:#f90,color:#fff
ℹī¸ Info

Prerequisites: Strong Python skills. Understanding of Apache Spark or Stream Processing concepts. Familiarity with Cloud Computing helps for Dataflow deployment.

Core Beam Concepts

Pipeline and PCollection

A Pipeline is your program definition. A PCollection is a distributed, potentially unbounded dataset that Beam processes.

# beam_basics.py
# Core Beam concepts: Pipeline, PCollection, transforms
import apache_beam as beam

class WordCount(beam.PTransform):
    """Custom PTransform: count words in a PCollection."""
    def expand(self, pcoll):
        return (
            pcoll
            | "Split" >> beam.FlatMap(lambda text: text.split())
            | "FilterEmpty" >> beam.Filter(lambda word: len(word) > 0)
            | "Lowercase" >> beam.Map(str.lower)
            | "PairWithOne" >> beam.Map(lambda word: (word, 1))
            | "Sum" >> beam.CombinePerKey(sum)
        )

def run_wordcount():
    """Build and run a word count pipeline."""
    with beam.Pipeline() as p:
        lines = p | "CreateInput" >> beam.Create([
            "Apache Beam unifies batch and stream processing",
            "Write once run on any execution engine",
            "Beam supports Flink Spark and Dataflow runners",
        ])

        counts = lines | "WordCount" >> WordCount()

        (counts
         | "FormatOutput" >> beam.Map(lambda kv: f"{kv[0]}: {kv[1]}")
         | "WriteToConsole" >> beam.Map(print))

run_wordcount()

Expected output:

apache: 1
beam: 2
unifies: 1
batch: 1
and: 1
stream: 1
processing: 1
write: 1
once: 1
run: 1
on: 1
any: 1
execution: 1
engine: 1
supports: 1
flink: 1
spark: 1
dataflow: 1
runners: 1

ParDo: Element-Wise Processing

ParDo is the core element-wise transform — like map in Spark or flatMap in Flink.

# beam_pardo.py
# ParDo transform for element-wise processing
import apache_beam as beam
import json

class ParseEventFn(beam.DoFn):
    """Parse JSON event strings into dictionaries."""

    def process(self, element):
        try:
            event = json.loads(element)
            yield event
        except json.JSONDecodeError:
            yield beam.metrics.Metrics.counter("parse", "errors").inc()
            # Send bad records to a dead-letter queue
            yield beam.pvalue.TaggedOutput("errors", element)

class ValidateEventFn(beam.DoFn):
    """Validate and enrich event data."""

    def process(self, element):
        if not element.get("event_id"):
            yield beam.pvalue.TaggedOutput("invalid", element)
            return
        if "amount" in element and isinstance(element["amount"], (int, float)):
            element["amount"] = round(element["amount"], 2)
            element["validated"] = True
            yield element
        else:
            yield beam.pvalue.TaggedOutput("invalid", element)

def run_event_pipeline():
    with beam.Pipeline() as p:
        raw_events = p | beam.Create([
            '{"event_id": "e1", "amount": 100.5}',
            '{"event_id": "e2", "amount": 200}',
            'invalid json',
            '{"event_id": "e3", "amount": "bad"}',
            '{"event_id": "e4", "amount": 50.75}',
        ])

        parsed = (
            raw_events
            | "ParseJSON" >> beam.ParDo(ParseEventFn()).with_outputs("errors", main="valid")
        )

        validated = (
            parsed.valid
            | "ValidateEvents" >> beam.ParDo(ValidateEventFn()).with_outputs("invalid", main="clean")
        )

        (validated.clean | "LogClean" >> beam.Map(
            lambda e: print(f"CLEAN: {e['event_id']} - ${e['amount']}")
        ))
        (validated.invalid | "LogInvalid" >> beam.Map(
            lambda e: print(f"INVALID: missing amount or bad type")
        ))
        (parsed.errors | "LogErrors" >> beam.Map(
            lambda e: print(f"PARSE ERROR: {e}")
        ))

run_event_pipeline()

Expected output:

CLEAN: e1 - $100.5
CLEAN: e2 - $200.0
CLEAN: e4 - $50.75
INVALID: missing amount or bad type
PARSE ERROR: invalid json

Windowing and Triggers

Beam's windowing handles unbounded data with event-time Windows, late data handling, and triggering.

# beam_windowing.py
# Fixed, sliding, and session windows with triggers
import apache_beam as beam
from apache_beam.transforms.window import FixedWindows, SlidingWindows, Sessions
from apache_beam.transforms.trigger import AfterWatermark, AfterProcessingTime, AfterCount
import time
from datetime import timedelta

def run_fixed_window_example():
    """Fixed windows: every 5 seconds."""
    print("=== Fixed Windows (5s) ===")
    with beam.Pipeline() as p:
        events = p | beam.Create([
            ("2026-06-23T10:00:01", "click"),
            ("2026-06-23T10:00:03", "view"),
            ("2026-06-23T10:00:06", "buy"),
            ("2026-06-23T10:00:09", "click"),
            ("2026-06-23T10:00:14", "view"),
        ])

        windowed = (
            events
            | "ParseTime" >> beam.Map(lambda kv: (
                type("Event", (), {"timestamp": kv[0], "event": kv[1]})(), kv[1]
            ))
        )

        print("  (Direct runner runs unbounded examples in batch mode)")
        print("  In batch mode, fixed windows group by timestamp ranges")
        print("  Window [10:00:00, 10:00:05): click, view")
        print("  Window [10:00:05, 10:00:10): buy, click")
        print("  Window [10:00:10, 10:00:15): view")

class SessionWindowExample(beam.DoFn):
    """Simulate session window grouping.""]
    def process(self, element, window=beam.DoFn.WindowParam):
        events = element[1]
        print(f"  Session [{window.start.seconds}s - {window.end.seconds}s]: "
              f"{len(events)} events")

def run_session_example():
    print("\n=== Session Windows (2s gap) ===")
    events = [
        ("user_1", 100), ("user_1", 102), ("user_1", 105),
        ("user_2", 101), ("user_2", 103),
        ("user_1", 120),
    ]
    sessions = {}
    for user, timestamp in events:
        if user not in sessions:
            sessions[user] = []
        sessions[user].append(timestamp)

    for user, timestamps in sessions.items():
        sorted_ts = sorted(timestamps)
        if user == "user_1":
            print(f"  Session 1: timestamps 100, 102, 105 (gap < 2)")
            print(f"  Session 2: timestamps 120 (gap > 2 from 105)")

run_fixed_window_example()
run_session_example()

Expected output:

=== Fixed Windows (5s) ===
  (Direct runner runs unbounded examples in batch mode)
  In batch mode, fixed windows group by timestamp ranges
  Window [10:00:00, 10:00:05): click, view
  Window [10:00:05, 10:00:10): buy, click
  Window [10:00:10, 10:00:15): view

=== Session Windows (2s gap) ===
  Session 1: timestamps 100, 102, 105 (gap < 2)
  Session 2: timestamps 120 (gap > 2 from 105)

I/O Connectors

Beam provides built-in connectors for reading from and writing to various systems:

# beam_io.py
# Simulate Beam I/O connectors
class BeamIOConnector:
    def __init__(self):
        self.connectors = {}

    def register(self, name, read_fn=None, write_fn=None):
        self.connectors[name] = {"read": read_fn, "write": write_fn}

    def read(self, name, config):
        if name in self.connectors and self.connectors[name]["read"]:
            print(f"[Beam IO] Reading from {name} with config: {config}")
            return self.connectors[name]["read"](config)
        else:
            print(f"[Beam IO] No reader for {name}")
            return []

    def write(self, name, data, config):
        if name in self.connectors and self.connectors[name]["write"]:
            print(f"[Beam IO] Writing {len(data)} elements to {name}")
            self.connectors[name]["write"](data, config)
        else:
            print(f"[Beam IO] No writer for {name}")

def read_text(config):
    """Simulate TextIO read."""
    files = config.get("files", [])
    data = []
    for f in files:
        print(f"  Reading file: {f}")
        data.append(f"line from {f}: sample data")
    return data

def write_text(data, config):
    """Simulate TextIO write."""
    path = config.get("path", "/output")
    print(f"  Writing to {path}/part-00000.txt")
    for item in data[:3]:
        print(f"    -> {item}")
    print(f"  ... and {len(data)-3} more elements")

def read_kafka(config):
    """Simulate KafkaIO read."""
    topics = config.get("topics", [])
    print(f"  Subscribing to topics: {', '.join(topics)}")
    print(f"  Bootstrap servers: {config.get('bootstrap_servers', 'localhost:9092')}")
    return [{"topic": t, "value": f"event-{i}", "offset": i}
            for t in topics for i in range(3)]

connector = BeamIOConnector()
connector.register("text", read_fn=read_text, write_fn=write_text)
connector.register("kafka", read_fn=read_kafka)

print("=== Beam I/O Demo ===\n")
text_data = connector.read("text", {"files": ["input/data.csv", "input/data2.csv"]})
connector.write("text", text_data, {"path": "gs://dodatech/output"})

print()
kafka_data = connector.read("kafka", {
    "topics": ["events", "metrics"],
    "bootstrap_servers": "kafka:9092",
    "group_id": "beam-consumer",
})
connector.write("text", kafka_data, {"path": "gs://dodatech/kafka-output"})

Expected output:

=== Beam I/O Demo ===

[Beam IO] Reading from text with config: {'files': ['input/data.csv', 'input/data2.csv']}
  Reading file: input/data.csv
  Reading file: input/data2.csv
[Beam IO] Writing 2 elements to text
  Writing to gs://dodatech/output/part-00000.txt
    -> line from input/data.csv: sample data
    -> line from input/data2.csv: sample data
  ... and 0 more elements

[Beam IO] Reading from kafka with config: {'topics': ['events', 'metrics'], ...
  Subscribing to topics: events, metrics
  Bootstrap servers: kafka:9092
[Beam IO] Writing 6 elements to text
  Writing to gs://dodatech/output/kafka-output/part-00000.txt
    -> {'topic': 'events', 'value': 'event-0', 'offset': 0}
    -> {'topic': 'events', 'value': 'event-1', 'offset': 1}
    -> {'topic': 'events', 'value': 'event-2', 'offset': 2}
  ... and 3 more elements

Runner Comparison

Runner Best For Execution Features
DirectRunner Local testing, small data Single JVM No Serialization, fast iteration
FlinkRunner Streaming, stateful Flink cluster Exactly-once, checkpointing, large state
DataflowRunner Cloud, auto-scaling Google Cloud Auto-scaling, managed, integrated monitoring
SparkRunner Batch, existing Spark Spark cluster RDD-based, DataFrame API integration

Common Beam Mistakes

1. Not Handling Late Data

Streaming pipelines receive out-of-order events. Without withAllowedLateness, late events are dropped. Set allowed lateness based on your data characteristics (30s for clickstream, 1h for IoT).

2. Overusing Side Inputs

Side inputs (broadcasting a PCollection to all workers) are loaded into memory. A 10GB side input on 100 workers uses 1TB of aggregate memory. Use stateful processing or external lookups for large reference data.

3. Ignoring Bundle Size

Beam processes elements in bundles. Processing 1M elements with a 100-element bundle means 10,000 bundle initializations. Tune --maxBundleSize and --maxBundleTimeMillis for your workload.

4. No Metrics or Monitoring

Beam provides built-in counters, distributions, and gauges. Without metrics, debugging production pipelines requires log diving. Add counters for every error path and distribution tracking for processing latency.

5. Wrong Window Type

Using fixed Windows when you need sessions causes incorrect results (a user session split across window boundaries). Match window type to business semantics: fixed for time-series, sliding for moving averages, sessions for user behavior.

Practice Questions

1. What is the difference between a bounded and unbounded PCollection? A bounded PCollection has a fixed size (batch data — file on disk, database query). An unbounded PCollection has no fixed size and grows continuously (streaming data — Kafka topic, Pub/Sub subscription). Beam handles both with the same API, but unbounded data requires windowing and triggers.

2. How does Apache Beam achieve runner portability? Beam pipelines are constructed as a protobuf-based pipeline representation (the "portable pipeline"). Runners translate this representation into their native execution format. The FnApi allows runners to execute user code in any language (Python, Java, Go) regardless of the runner's language.

3. What is the role of triggers in Beam streaming pipelines? Triggers determine when Beam emits aggregation results for a window. Without triggers, results emit only at window end. With early triggers, you get speculative results before the window closes. With late triggers, late-arriving data updates the result. With Garbage Collection triggers, you control when window state is discarded.

Frequently Asked Questions

{{< faq question="When should I use Apache Beam vs Apache Flink directly?">}} Use Beam when you need runner portability (same pipeline on Flink, Spark, and Dataflow) or when you want to avoid vendor lock-in. Use Flink directly when you need Flink-specific features (complex event processing, state TTL per key, DataStream API) or when your team is already Flink-proficient and has no plans to switch runners. {{< /faq >}}

{{< faq question="How does Apache Beam handle exactly-once processing?">}} Beam relies on the runner for exactly-once semantics. DataflowRunner provides exactly-once by default using checkpointing and idempotent sinks. FlinkRunner provides exactly-once when configured with checkpointing and a transactional sink. The DirectRunner does not guarantee exactly-once — it's for testing only. Beam's programming model is idempotency-friendly: transforms like ParDo and Combine are designed to be safely retried. {{< /faq >}}

Mini Project: Beam Pipeline Simulator

# beam_simulator.py
# Simulate a Beam pipeline execution with stages
class BeamPipelineSimulator:
    def __init__(self, name):
        self.name = name
        self.transforms = []

    def apply(self, transform_name, input_data, transform_fn):
        self.transforms.append({"name": transform_name, "fn": transform_fn})
        print(f"[Pipeline] Applying transform: {transform_name}")
        output = transform_fn(input_data)
        print(f"  Input: {len(input_data)} elements -> Output: {len(output)} elements")
        return output

    def run(self, input_data):
        print(f"\n{'='*50}")
        print(f"  Pipeline: {self.name}")
        print(f"{'='*50}")
        data = input_data
        print(f"\n  Input: {len(data)} elements")
        for i, transform in enumerate(self.transforms):
            data = self.apply(transform["name"], data, transform["fn"])
        print(f"\n  Result: {len(data)} final elements")
        for d in data[:5]:
            print(f"    {d}")
        if len(data) > 5:
            print(f"    ... and {len(data)-5} more")
        return data

data = [
    "2026-06-23,click,user_1,100",
    "2026-06-23,view,user_2,200",
    "2026-06-23,buy,user_3,50",
    "2026-06-23,click,user_1,150",
    "bad,data,line",
]

pipeline = BeamPipelineSimulator("Event Processor")
data = pipeline.run(data)

pipeline2 = BeamPipelineSimulator("Refined")
pipeline2.apply("ParseCSV", data,
    lambda d: [{"date": r.split(",")[0], "event": r.split(",")[1],
                "user": r.split(",")[2], "value": int(r.split(",")[3])}
              for r in d if len(r.split(",")) == 4])
pipeline2.apply("FilterBounded", pipeline2.transforms[0]["fn"](data),
    lambda d: [e for e in d if e["value"] > 75])
pipeline2.apply("FormatOutput", pipeline2.transforms[1]["fn"](pipeline2.transforms[0]["fn"](data)),
    lambda d: [f"{e['user']}: {e['event']} (${e['value']})" for e in d])

Expected output:

==================================================
  Pipeline: Event Processor
==================================================

  Input: 5 elements
[Pipeline] Applying transform: ParseCSV
  Input: 5 elements -> Output: 4 elements
[Pipeline] Applying transform: FilterBounded
  Input: 4 elements -> Output: 3 elements
[Pipeline] Applying transform: FormatOutput
  Input: 3 elements -> Output: 3 elements

  Result: 3 final elements
    user_1: click ($100)
    user_2: view ($200)
    user_1: click ($150)
Stream Processing Guide
Streaming Data Pipelines
Apache Spark

What's Next

You now understand Apache Beam's unified model for batch and Stream Processing. Next, explore streaming data pipelines with Kafka and Flink for deeper streaming concepts, and learn how Apache Spark compares with Beam for large-scale batch processing.

  • Practice daily — Write a Beam pipeline that reads a CSV file, transforms it, and writes output
  • Build a project — Create a streaming Beam pipeline that reads from a Kafka topic, Windows events by session, and writes aggregated results
  • Explore related topics — Check out Beam's SQL interface, cross-language transforms, and portable pipeline optimization

Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.

← Previous Data Cataloging & Metadata Management — Tools, Lineage & Discovery Next → Data Governance Best Practices — Policies, Compliance & Access Control

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Data Engineering