Core Programming Model

Module Duration: 3-4 hours Focus: Fundamental Beam abstractions and transformations Prerequisites: Java or Python, basic data processing concepts

Overview

Apache Beam’s core programming model provides a unified abstraction for both batch and streaming data processing. This module covers the fundamental concepts that make Beam portable across different execution engines.

Key Concepts

PCollection: Immutable, distributed dataset
Pipeline: DAG of transformations
PTransform: Data transformation operation
ParDo: Parallel element-wise processing
DoFn: User-defined function for ParDo

Understanding PCollections

A PCollection (Parallel Collection) represents a distributed dataset that your Beam pipeline operates on. PCollections are immutable and can be bounded (batch) or unbounded (streaming).

PCollection Characteristics

Immutability: Once created, a PCollection cannot be modified. Transformations create new PCollections. Distributed: Elements are distributed across multiple workers for parallel processing. Timestamped: Each element has an associated timestamp (critical for streaming). Windowed: Elements are organized into windows for grouping operations.

Creating PCollections

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# Create pipeline
with beam.Pipeline(options=PipelineOptions()) as pipeline:

    # From in-memory data
    numbers = pipeline | "Create Numbers" >> beam.Create([1, 2, 3, 4, 5])

    # From text file
    lines = pipeline | "Read File" >> beam.io.ReadFromText('input.txt')

    # From multiple files with pattern
    logs = pipeline | "Read Logs" >> beam.io.ReadFromText('logs/*.log')

    # From Kafka (unbounded)
    events = pipeline | "Read Kafka" >> beam.io.ReadFromKafka(
        consumer_config={'bootstrap.servers': 'localhost:9092'},
        topics=['events']
    )

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.transforms.Create;

// Create pipeline
Pipeline pipeline = Pipeline.create(options);

// From in-memory data
PCollection<Integer> numbers = pipeline.apply(
    "Create Numbers",
    Create.of(1, 2, 3, 4, 5)
);

// From text file
PCollection<String> lines = pipeline.apply(
    "Read File",
    TextIO.read().from("input.txt")
);

// From multiple files with pattern
PCollection<String> logs = pipeline.apply(
    "Read Logs",
    TextIO.read().from("logs/*.log")
);

// From Kafka (unbounded)
PCollection<KafkaRecord<String, String>> events = pipeline.apply(
    "Read Kafka",
    KafkaIO.<String, String>read()
        .withBootstrapServers("localhost:9092")
        .withTopic("events")
);

PCollection Types

Bounded PCollection: Finite dataset with a known end (batch processing)

Reading from files
Database queries
Fixed in-memory collections

Unbounded PCollection: Infinite dataset that continues indefinitely (streaming)

Kafka topics
Pub/Sub subscriptions
Real-time sensor data

Pipeline Construction

A Pipeline represents the entire data processing workflow as a Directed Acyclic Graph (DAG) of transformations.

Pipeline Creation and Configuration

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# Basic pipeline
pipeline = beam.Pipeline()

# Pipeline with options
options = PipelineOptions([
    '--runner=DirectRunner',
    '--project=my-project',
    '--region=us-central1',
    '--temp_location=gs://my-bucket/temp',
    '--job_name=word-count-job'
])

pipeline = beam.Pipeline(options=options)

# Context manager (recommended)
with beam.Pipeline(options=options) as p:
    # Pipeline operations
    result = (p
        | "Read" >> beam.io.ReadFromText('input.txt')
        | "Process" >> beam.Map(lambda x: x.upper())
        | "Write" >> beam.io.WriteToText('output.txt')
    )
    # Pipeline automatically runs and waits at end of context

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;

// Basic pipeline
Pipeline pipeline = Pipeline.create();

// Pipeline with options
PipelineOptions options = PipelineOptionsFactory.fromArgs(args)
    .withValidation()
    .as(PipelineOptions.class);

Pipeline pipeline = Pipeline.create(options);

// Build pipeline
pipeline
    .apply("Read", TextIO.read().from("input.txt"))
    .apply("Process", MapElements.via(new SimpleFunction<String, String>() {
        public String apply(String input) {
            return input.toUpperCase();
        }
    }))
    .apply("Write", TextIO.write().to("output.txt"));

// Run pipeline
PipelineResult result = pipeline.run();
result.waitUntilFinish();

Pipeline Options

from apache_beam.options.pipeline_options import PipelineOptions

class MyOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_argument('--input', required=True)
        parser.add_argument('--output', required=True)
        parser.add_argument('--max_workers', type=int, default=10)

options = MyOptions([
    '--input=gs://bucket/input',
    '--output=gs://bucket/output',
    '--max_workers=20',
    '--runner=DataflowRunner'
])

with beam.Pipeline(options=options) as pipeline:
    input_path = options.input
    output_path = options.output
    # Use options in pipeline

import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;

public interface MyOptions extends PipelineOptions {
    @Description("Input file path")
    String getInput();
    void setInput(String value);

    @Description("Output file path")
    String getOutput();
    void setOutput(String value);

    @Description("Maximum number of workers")
    @Default.Integer(10)
    Integer getMaxWorkers();
    void setMaxWorkers(Integer value);
}

MyOptions options = PipelineOptionsFactory.fromArgs(args)
    .withValidation()
    .as(MyOptions.class);

Pipeline pipeline = Pipeline.create(options);

PTransforms: Data Transformations

PTransforms are operations that transform PCollections. They represent the nodes in your pipeline DAG.

Core Transform Types

Element-wise: Process each element independently (Map, FlatMap, Filter) Aggregating: Combine elements (GroupByKey, Combine, Count) Composite: Combine multiple transforms into reusable units

Map Transform

Apply a 1:1 function to each element.

import apache_beam as beam

def double(x):
    return x * 2

with beam.Pipeline() as pipeline:
    # Using Map with function
    doubled = (pipeline
        | beam.Create([1, 2, 3, 4, 5])
        | beam.Map(double)
    )

    # Using Map with lambda
    squared = (pipeline
        | beam.Create([1, 2, 3, 4, 5])
        | beam.Map(lambda x: x * x)
    )

    # Map with additional arguments
    multiplied = (pipeline
        | beam.Create([1, 2, 3, 4, 5])
        | beam.Map(lambda x, factor: x * factor, factor=10)
    )

import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

PCollection<Integer> doubled = numbers.apply(
    "Double",
    MapElements
        .into(TypeDescriptors.integers())
        .via(x -> x * 2)
);

PCollection<Integer> squared = numbers.apply(
    "Square",
    MapElements
        .into(TypeDescriptors.integers())
        .via(x -> x * x)
);

FlatMap Transform

Apply a 1:N function (each input produces zero or more outputs).

import apache_beam as beam

def split_words(text):
    return text.split()

with beam.Pipeline() as pipeline:
    words = (pipeline
        | beam.Create(['Hello World', 'Apache Beam', 'Data Processing'])
        | beam.FlatMap(split_words)
    )
    # Output: ['Hello', 'World', 'Apache', 'Beam', 'Data', 'Processing']

    # FlatMap with lambda
    chars = (pipeline
        | beam.Create(['abc', 'def'])
        | beam.FlatMap(lambda x: list(x))
    )
    # Output: ['a', 'b', 'c', 'd', 'e', 'f']

    # FlatMap can filter by returning empty list
    evens = (pipeline
        | beam.Create([1, 2, 3, 4, 5])
        | beam.FlatMap(lambda x: [x] if x % 2 == 0 else [])
    )

import org.apache.beam.sdk.transforms.FlatMapElements;

PCollection<String> words = lines.apply(
    "Split Words",
    FlatMapElements
        .into(TypeDescriptors.strings())
        .via(line -> Arrays.asList(line.split("\\s+")))
);

PCollection<Character> chars = words.apply(
    "Split Chars",
    FlatMapElements
        .into(TypeDescriptors.chars())
        .via(word -> word.chars()
            .mapToObj(c -> (char) c)
            .collect(Collectors.toList()))
);

Filter Transform

Keep only elements matching a predicate.

import apache_beam as beam

with beam.Pipeline() as pipeline:
    even_numbers = (pipeline
        | beam.Create([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
        | beam.Filter(lambda x: x % 2 == 0)
    )

    # Filter with named function
    def is_valid(record):
        return record.get('status') == 'active' and record.get('value') > 0

    valid_records = (pipeline
        | beam.Create([
            {'status': 'active', 'value': 100},
            {'status': 'inactive', 'value': 50},
            {'status': 'active', 'value': -10}
        ])
        | beam.Filter(is_valid)
    )

import org.apache.beam.sdk.transforms.Filter;

PCollection<Integer> evenNumbers = numbers.apply(
    "Filter Evens",
    Filter.by(x -> x % 2 == 0)
);

PCollection<String> longLines = lines.apply(
    "Filter Long Lines",
    Filter.by(line -> line.length() > 100)
);

ParDo and DoFn

ParDo is the most fundamental and flexible transform in Beam. It allows you to implement custom processing logic through DoFn (Do Function).

Basic DoFn

import apache_beam as beam
from apache_beam.transforms import DoFn

class ExtractWordsFn(DoFn):
    def process(self, element):
        """Process a single element.

        Args:
            element: Input element

        Yields:
            Processed outputs
        """
        words = element.split()
        for word in words:
            yield word.lower()

# Using DoFn
with beam.Pipeline() as pipeline:
    words = (pipeline
        | beam.Create(['Hello World', 'Apache Beam'])
        | beam.ParDo(ExtractWordsFn())
    )

import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;

static class ExtractWordsFn extends DoFn<String, String> {
    @ProcessElement
    public void processElement(@Element String element, OutputReceiver<String> out) {
        String[] words = element.split("\\s+");
        for (String word : words) {
            out.output(word.toLowerCase());
        }
    }
}

PCollection<String> words = lines.apply(
    "Extract Words",
    ParDo.of(new ExtractWordsFn())
);

DoFn Lifecycle Methods

DoFn provides lifecycle methods for setup and teardown operations.

import apache_beam as beam
from apache_beam.transforms import DoFn
import logging

class EnrichDataFn(DoFn):
    def __init__(self, api_endpoint):
        self.api_endpoint = api_endpoint
        self.client = None

    def setup(self):
        """Called once per worker before processing."""
        logging.info("Setting up worker resources")
        import requests
        self.client = requests.Session()
        self.cache = {}

    def start_bundle(self):
        """Called at the start of each bundle."""
        logging.info("Starting new bundle")
        self.bundle_count = 0

    def process(self, element):
        """Process each element."""
        self.bundle_count += 1

        # Use cache to avoid repeated API calls
        key = element['id']
        if key not in self.cache:
            response = self.client.get(f"{self.api_endpoint}/{key}")
            self.cache[key] = response.json()

        enriched = {**element, 'metadata': self.cache[key]}
        yield enriched

    def finish_bundle(self):
        """Called at the end of each bundle."""
        logging.info(f"Finished bundle with {self.bundle_count} elements")

    def teardown(self):
        """Called once per worker after all processing."""
        logging.info("Tearing down worker resources")
        if self.client:
            self.client.close()

with beam.Pipeline() as pipeline:
    enriched = (pipeline
        | beam.Create([{'id': '1'}, {'id': '2'}])
        | beam.ParDo(EnrichDataFn('https://api.example.com'))
    )

import org.apache.beam.sdk.transforms.DoFn;

static class EnrichDataFn extends DoFn<KV<String, String>, KV<String, String>> {
    private final String apiEndpoint;
    private transient HttpClient client;
    private transient Map<String, String> cache;

    public EnrichDataFn(String apiEndpoint) {
        this.apiEndpoint = apiEndpoint;
    }

    @Setup
    public void setup() {
        LOG.info("Setting up worker resources");
        client = HttpClient.newHttpClient();
        cache = new HashMap<>();
    }

    @StartBundle
    public void startBundle() {
        LOG.info("Starting new bundle");
    }

    @ProcessElement
    public void processElement(@Element KV<String, String> element,
                               OutputReceiver<KV<String, String>> out) {
        String key = element.getKey();

        if (!cache.containsKey(key)) {
            String metadata = fetchMetadata(key);
            cache.put(key, metadata);
        }

        String enriched = element.getValue() + ":" + cache.get(key);
        out.output(KV.of(key, enriched));
    }

    @FinishBundle
    public void finishBundle() {
        LOG.info("Finished bundle");
    }

    @Teardown
    public void teardown() {
        LOG.info("Tearing down worker resources");
    }

    private String fetchMetadata(String key) {
        // API call implementation
        return "metadata";
    }
}

Multiple Outputs (Side Outputs)

DoFn can emit elements to multiple output PCollections using tagged outputs.

import apache_beam as beam
from apache_beam.transforms import DoFn

class SplitDataFn(DoFn):
    # Define output tags
    OUTPUT_TAG_VALID = 'valid'
    OUTPUT_TAG_INVALID = 'invalid'
    OUTPUT_TAG_WARNING = 'warning'

    def process(self, element):
        value = element.get('value', 0)

        if value < 0:
            # Emit to invalid output
            yield beam.pvalue.TaggedOutput(self.OUTPUT_TAG_INVALID, element)
        elif value > 1000:
            # Emit to warning output
            yield beam.pvalue.TaggedOutput(self.OUTPUT_TAG_WARNING, element)
        else:
            # Main output (no tag)
            yield element

with beam.Pipeline() as pipeline:
    data = pipeline | beam.Create([
        {'id': 1, 'value': 100},
        {'id': 2, 'value': -50},
        {'id': 3, 'value': 2000},
        {'id': 4, 'value': 500}
    ])

    # Apply DoFn with side outputs
    results = data | beam.ParDo(SplitDataFn()).with_outputs(
        SplitDataFn.OUTPUT_TAG_INVALID,
        SplitDataFn.OUTPUT_TAG_WARNING,
        main='valid'
    )

    # Access outputs
    valid = results.valid
    invalid = results.invalid
    warning = results.warning

    # Process each output separately
    valid | "Write Valid" >> beam.io.WriteToText('valid.txt')
    invalid | "Write Invalid" >> beam.io.WriteToText('invalid.txt')
    warning | "Write Warning" >> beam.io.WriteToText('warning.txt')

import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.sdk.values.PCollectionTuple;

// Define output tags
final TupleTag<Record> validTag = new TupleTag<Record>(){};
final TupleTag<Record> invalidTag = new TupleTag<Record>(){};
final TupleTag<Record> warningTag = new TupleTag<Record>(){};

static class SplitDataFn extends DoFn<Record, Record> {
    private final TupleTag<Record> invalidTag;
    private final TupleTag<Record> warningTag;

    public SplitDataFn(TupleTag<Record> invalidTag, TupleTag<Record> warningTag) {
        this.invalidTag = invalidTag;
        this.warningTag = warningTag;
    }

    @ProcessElement
    public void processElement(@Element Record element,
                               MultiOutputReceiver out) {
        int value = element.getValue();

        if (value < 0) {
            out.get(invalidTag).output(element);
        } else if (value > 1000) {
            out.get(warningTag).output(element);
        } else {
            out.output(element);
        }
    }
}

PCollectionTuple results = data.apply(
    "Split Data",
    ParDo.of(new SplitDataFn(invalidTag, warningTag))
        .withOutputTags(validTag, TupleTagList.of(invalidTag).and(warningTag))
);

PCollection<Record> valid = results.get(validTag);
PCollection<Record> invalid = results.get(invalidTag);
PCollection<Record> warning = results.get(warningTag);

Side Inputs

Side inputs allow you to use additional data alongside the main input in a ParDo.

import apache_beam as beam
from apache_beam.transforms import DoFn

class EnrichWithDictFn(DoFn):
    def process(self, element, lookup_dict):
        """Enrich element using side input dictionary.

        Args:
            element: Main input element
            lookup_dict: Side input (passed as iterator)
        """
        # Side input is passed as iterator, convert to dict
        lookup = {k: v for d in lookup_dict for k, v in d.items()}

        key = element['id']
        enriched_value = lookup.get(key, 'unknown')

        yield {
            **element,
            'enriched': enriched_value
        }

with beam.Pipeline() as pipeline:
    # Main input
    main_data = pipeline | "Main Data" >> beam.Create([
        {'id': 'A', 'value': 100},
        {'id': 'B', 'value': 200},
        {'id': 'C', 'value': 300}
    ])

    # Side input (reference data)
    lookup_data = pipeline | "Lookup Data" >> beam.Create([
        {'A': 'Alpha', 'B': 'Beta', 'C': 'Gamma'}
    ])

    # Use side input in ParDo
    enriched = (main_data
        | beam.ParDo(EnrichWithDictFn(), lookup_dict=beam.pvalue.AsIter(lookup_data))
    )

    enriched | beam.Map(print)

import org.apache.beam.sdk.values.PCollectionView;

static class EnrichWithDictFn extends DoFn<KV<String, Integer>, KV<String, String>> {
    private final PCollectionView<Map<String, String>> lookupView;

    public EnrichWithDictFn(PCollectionView<Map<String, String>> lookupView) {
        this.lookupView = lookupView;
    }

    @ProcessElement
    public void processElement(@Element KV<String, Integer> element,
                               OutputReceiver<KV<String, String>> out,
                               ProcessContext c) {
        Map<String, String> lookup = c.sideInput(lookupView);

        String key = element.getKey();
        String enrichedValue = lookup.getOrDefault(key, "unknown");

        out.output(KV.of(key, enrichedValue));
    }
}

// Create side input view
PCollectionView<Map<String, String>> lookupView =
    lookupData.apply(View.asMap());

// Use side input
PCollection<KV<String, String>> enriched = mainData.apply(
    "Enrich",
    ParDo.of(new EnrichWithDictFn(lookupView))
        .withSideInputs(lookupView)
);

Composite Transforms

Composite transforms encapsulate multiple transforms into reusable components.

Creating Composite Transforms

import apache_beam as beam
from apache_beam.transforms import PTransform

class CountWords(PTransform):
    """Composite transform to count word frequencies."""

    def expand(self, pcoll):
        return (pcoll
            | "Split into words" >> beam.FlatMap(lambda x: x.split())
            | "Lowercase" >> beam.Map(str.lower)
            | "Filter empty" >> beam.Filter(lambda x: len(x) > 0)
            | "Pair with 1" >> beam.Map(lambda x: (x, 1))
            | "Group by key" >> beam.GroupByKey()
            | "Sum counts" >> beam.MapTuple(lambda word, ones: (word, sum(ones)))
        )

class FilterAndTransform(PTransform):
    """Composite transform with parameters."""

    def __init__(self, min_value, transform_fn):
        super().__init__()
        self.min_value = min_value
        self.transform_fn = transform_fn

    def expand(self, pcoll):
        return (pcoll
            | "Filter" >> beam.Filter(lambda x: x >= self.min_value)
            | "Transform" >> beam.Map(self.transform_fn)
        )

# Using composite transforms
with beam.Pipeline() as pipeline:
    lines = pipeline | beam.Create([
        'Hello World',
        'Apache Beam',
        'Data Processing'
    ])

    # Use CountWords composite transform
    word_counts = lines | CountWords()

    # Use parameterized composite transform
    numbers = pipeline | beam.Create([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
    result = numbers | FilterAndTransform(5, lambda x: x * x)

import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;

public class CountWords extends PTransform<PCollection<String>, PCollection<KV<String, Long>>> {

    @Override
    public PCollection<KV<String, Long>> expand(PCollection<String> lines) {
        return lines
            .apply("Split into words",
                FlatMapElements.into(TypeDescriptors.strings())
                    .via(line -> Arrays.asList(line.split("\\s+"))))
            .apply("Lowercase",
                MapElements.into(TypeDescriptors.strings())
                    .via(String::toLowerCase))
            .apply("Filter empty",
                Filter.by(word -> !word.isEmpty()))
            .apply("Count",
                Count.perElement());
    }
}

public class FilterAndTransform extends PTransform<PCollection<Integer>, PCollection<Integer>> {
    private final int minValue;
    private final SerializableFunction<Integer, Integer> transformFn;

    public FilterAndTransform(int minValue, SerializableFunction<Integer, Integer> transformFn) {
        this.minValue = minValue;
        this.transformFn = transformFn;
    }

    @Override
    public PCollection<Integer> expand(PCollection<Integer> input) {
        return input
            .apply("Filter", Filter.by(x -> x >= minValue))
            .apply("Transform", MapElements.into(TypeDescriptors.integers()).via(transformFn));
    }
}

// Usage
PCollection<KV<String, Long>> wordCounts = lines.apply(new CountWords());
PCollection<Integer> result = numbers.apply(new FilterAndTransform(5, x -> x * x));

Aggregation Operations

Beam provides several built-in transforms for aggregating data.

GroupByKey

Groups elements by key. Requires input PCollection of key-value pairs.

import apache_beam as beam

with beam.Pipeline() as pipeline:
    # Create key-value pairs
    kv_pairs = pipeline | beam.Create([
        ('cat', 'meow'),
        ('dog', 'woof'),
        ('cat', 'purr'),
        ('dog', 'bark'),
        ('cat', 'hiss')
    ])

    # Group by key
    grouped = kv_pairs | beam.GroupByKey()
    # Result: [('cat', ['meow', 'purr', 'hiss']), ('dog', ['woof', 'bark'])]

    # Process grouped results
    counts = grouped | beam.MapTuple(lambda key, values: (key, len(list(values))))

import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.values.KV;

PCollection<KV<String, String>> kvPairs = pipeline.apply(
    Create.of(
        KV.of("cat", "meow"),
        KV.of("dog", "woof"),
        KV.of("cat", "purr"),
        KV.of("dog", "bark"),
        KV.of("cat", "hiss")
    )
);

PCollection<KV<String, Iterable<String>>> grouped =
    kvPairs.apply(GroupByKey.create());

PCollection<KV<String, Integer>> counts = grouped.apply(
    MapElements.into(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers()))
        .via(kv -> KV.of(kv.getKey(), Iterables.size(kv.getValue())))
);

Combine

Efficiently aggregates all values for each key using associative and commutative operations.

import apache_beam as beam
from apache_beam.transforms import CombineFn

# Using built-in combiners
with beam.Pipeline() as pipeline:
    numbers = pipeline | beam.Create([1, 2, 3, 4, 5])

    # Sum all elements
    total = numbers | beam.CombineGlobally(sum)

    # Mean
    average = numbers | beam.CombineGlobally(beam.combiners.MeanCombineFn())

    # Count
    count = numbers | beam.combiners.Count.Globally()

# Custom CombineFn
class AverageFn(CombineFn):
    def create_accumulator(self):
        return (0.0, 0)  # (sum, count)

    def add_input(self, accumulator, input):
        sum_val, count = accumulator
        return (sum_val + input, count + 1)

    def merge_accumulators(self, accumulators):
        sums, counts = zip(*accumulators)
        return (sum(sums), sum(counts))

    def extract_output(self, accumulator):
        sum_val, count = accumulator
        return sum_val / count if count > 0 else 0

# Per-key combine
with beam.Pipeline() as pipeline:
    kv_pairs = pipeline | beam.Create([
        ('A', 1), ('B', 2), ('A', 3), ('B', 4), ('A', 5)
    ])

    # Sum per key
    sums = kv_pairs | beam.CombinePerKey(sum)
    # Result: [('A', 9), ('B', 6)]

    # Average per key
    averages = kv_pairs | beam.CombinePerKey(AverageFn())

import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.transforms.Mean;

// Using built-in combiners
PCollection<Integer> total = numbers.apply(Sum.integersGlobally());
PCollection<Double> average = numbers.apply(Mean.<Integer>globally());
PCollection<Long> count = numbers.apply(Count.globally());

// Custom CombineFn
public static class AverageFn extends CombineFn<Integer, AverageFn.Accum, Double> {

    public static class Accum {
        double sum = 0.0;
        int count = 0;
    }

    @Override
    public Accum createAccumulator() {
        return new Accum();
    }

    @Override
    public Accum addInput(Accum accum, Integer input) {
        accum.sum += input;
        accum.count++;
        return accum;
    }

    @Override
    public Accum mergeAccumulators(Iterable<Accum> accums) {
        Accum merged = createAccumulator();
        for (Accum accum : accums) {
            merged.sum += accum.sum;
            merged.count += accum.count;
        }
        return merged;
    }

    @Override
    public Double extractOutput(Accum accum) {
        return accum.count > 0 ? accum.sum / accum.count : 0.0;
    }
}

// Per-key combine
PCollection<KV<String, Integer>> sums =
    kvPairs.apply(Sum.integersPerKey());

PCollection<KV<String, Double>> averages =
    kvPairs.apply(Combine.perKey(new AverageFn()));

CoGroupByKey

Joins multiple PCollections with the same key type.

import apache_beam as beam

with beam.Pipeline() as pipeline:
    # Email addresses
    emails = pipeline | "Create Emails" >> beam.Create([
        ('alice', 'alice@example.com'),
        ('bob', 'bob@example.com')
    ])

    # Phone numbers
    phones = pipeline | "Create Phones" >> beam.Create([
        ('alice', '555-1234'),
        ('bob', '555-5678'),
        ('bob', '555-9999')
    ])

    # CoGroup
    joined = ({'emails': emails, 'phones': phones}
        | beam.CoGroupByKey()
    )

    # Process joined data
    def format_contact(key_value):
        name, data = key_value
        emails = list(data['emails'])
        phones = list(data['phones'])
        return f"{name}: {emails}, {phones}"

    result = joined | beam.Map(format_contact)
    # Result: ['alice: [alice@example.com], [555-1234]',
    #          'bob: [bob@example.com], [555-5678, 555-9999]']

import org.apache.beam.sdk.transforms.join.CoGroupByKey;
import org.apache.beam.sdk.transforms.join.KeyedPCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;

final TupleTag<String> emailsTag = new TupleTag<>();
final TupleTag<String> phonesTag = new TupleTag<>();

PCollection<KV<String, CoGbkResult>> joined =
    KeyedPCollectionTuple.of(emailsTag, emails)
        .and(phonesTag, phones)
        .apply(CoGroupByKey.create());

PCollection<String> result = joined.apply(
    MapElements.into(TypeDescriptors.strings())
        .via(kv -> {
            String name = kv.getKey();
            CoGbkResult data = kv.getValue();

            Iterable<String> emailList = data.getAll(emailsTag);
            Iterable<String> phoneList = data.getAll(phonesTag);

            return name + ": " + emailList + ", " + phoneList;
        })
);

Real-World Use Cases

Log Processing Pipeline

import apache_beam as beam
from apache_beam.transforms import DoFn
import json
import re
from datetime import datetime

class ParseLogFn(DoFn):
    LOG_PATTERN = re.compile(
        r'(?P<ip>\S+) - - \[(?P<timestamp>.*?)\] "(?P<method>\S+) (?P<path>\S+) (?P<protocol>\S+)" (?P<status>\d+) (?P<size>\d+)'
    )

    def process(self, element):
        match = self.LOG_PATTERN.match(element)
        if match:
            yield {
                'ip': match.group('ip'),
                'timestamp': match.group('timestamp'),
                'method': match.group('method'),
                'path': match.group('path'),
                'status': int(match.group('status')),
                'size': int(match.group('size'))
            }

class DetectAnomaliesFn(DoFn):
    def process(self, element):
        ip, requests = element
        request_list = list(requests)

        # Anomaly: More than 100 requests from single IP
        if len(request_list) > 100:
            yield {
                'type': 'high_volume',
                'ip': ip,
                'count': len(request_list)
            }

        # Anomaly: High error rate
        errors = sum(1 for r in request_list if r['status'] >= 400)
        if errors > len(request_list) * 0.5:
            yield {
                'type': 'high_error_rate',
                'ip': ip,
                'error_rate': errors / len(request_list)
            }

def run_log_pipeline(input_path, output_path):
    with beam.Pipeline() as pipeline:
        logs = pipeline | "Read Logs" >> beam.io.ReadFromText(input_path)

        parsed = logs | "Parse Logs" >> beam.ParDo(ParseLogFn())

        # Count requests per IP
        ip_counts = (parsed
            | "Extract IP" >> beam.Map(lambda x: (x['ip'], x))
            | "Group by IP" >> beam.GroupByKey()
            | "Detect Anomalies" >> beam.ParDo(DetectAnomaliesFn())
        )

        # Count status codes
        status_counts = (parsed
            | "Extract Status" >> beam.Map(lambda x: (x['status'], 1))
            | "Count per Status" >> beam.CombinePerKey(sum)
        )

        # Write results
        ip_counts | "Write Anomalies" >> beam.io.WriteToText(f"{output_path}/anomalies")
        status_counts | "Write Status" >> beam.io.WriteToText(f"{output_path}/status_counts")

static class ParseLogFn extends DoFn<String, LogEntry> {
    private static final Pattern LOG_PATTERN = Pattern.compile(
        "(\\S+) - - \\[(.+?)\\] \"(\\S+) (\\S+) (\\S+)\" (\\d+) (\\d+)"
    );

    @ProcessElement
    public void processElement(@Element String element, OutputReceiver<LogEntry> out) {
        Matcher matcher = LOG_PATTERN.matcher(element);
        if (matcher.matches()) {
            LogEntry entry = new LogEntry(
                matcher.group(1),  // ip
                matcher.group(2),  // timestamp
                matcher.group(3),  // method
                matcher.group(4),  // path
                Integer.parseInt(matcher.group(6)),  // status
                Integer.parseInt(matcher.group(7))   // size
            );
            out.output(entry);
        }
    }
}

static class DetectAnomaliesFn extends DoFn<KV<String, Iterable<LogEntry>>, Anomaly> {
    @ProcessElement
    public void processElement(@Element KV<String, Iterable<LogEntry>> element,
                               OutputReceiver<Anomaly> out) {
        String ip = element.getKey();
        List<LogEntry> requests = Lists.newArrayList(element.getValue());

        if (requests.size() > 100) {
            out.output(new Anomaly("high_volume", ip, requests.size()));
        }

        long errors = requests.stream()
            .filter(r -> r.getStatus() >= 400)
            .count();

        if (errors > requests.size() * 0.5) {
            out.output(new Anomaly("high_error_rate", ip,
                (double) errors / requests.size()));
        }
    }
}

ETL Pipeline

import apache_beam as beam
from apache_beam.transforms import DoFn
import json

class ValidateRecordFn(DoFn):
    def process(self, element):
        try:
            record = json.loads(element)

            # Validation rules
            required_fields = ['id', 'name', 'email', 'created_at']
            if all(field in record for field in required_fields):
                if '@' in record['email']:
                    yield record
                else:
                    yield beam.pvalue.TaggedOutput('invalid_email', record)
            else:
                yield beam.pvalue.TaggedOutput('missing_fields', record)
        except json.JSONDecodeError:
            yield beam.pvalue.TaggedOutput('parse_error', element)

class TransformRecordFn(DoFn):
    def process(self, element):
        from datetime import datetime

        transformed = {
            'user_id': element['id'],
            'full_name': element['name'].upper(),
            'email': element['email'].lower(),
            'signup_date': datetime.fromisoformat(element['created_at']).date().isoformat(),
            'processed_at': datetime.utcnow().isoformat()
        }

        # Add computed fields
        transformed['email_domain'] = element['email'].split('@')[1]

        yield transformed

def run_etl_pipeline():
    with beam.Pipeline() as pipeline:
        raw_data = pipeline | "Read JSON" >> beam.io.ReadFromText('input.json')

        # Validate
        validation_results = (raw_data
            | "Validate" >> beam.ParDo(ValidateRecordFn()).with_outputs(
                'invalid_email',
                'missing_fields',
                'parse_error',
                main='valid'
            )
        )

        # Transform valid records
        transformed = (validation_results.valid
            | "Transform" >> beam.ParDo(TransformRecordFn())
        )

        # Aggregate by email domain
        domain_counts = (transformed
            | "Extract Domain" >> beam.Map(lambda x: (x['email_domain'], 1))
            | "Count per Domain" >> beam.CombinePerKey(sum)
        )

        # Write outputs
        transformed | "Write Valid" >> beam.io.WriteToText('output/valid')
        validation_results.invalid_email | "Write Invalid Email" >> beam.io.WriteToText('output/invalid_email')
        validation_results.missing_fields | "Write Missing" >> beam.io.WriteToText('output/missing_fields')
        domain_counts | "Write Domains" >> beam.io.WriteToText('output/domain_counts')

Best Practices

Performance Optimization

1. Minimize data serialization

# Bad: Multiple maps with serialization overhead
result = (data
    | beam.Map(lambda x: x.upper())
    | beam.Map(lambda x: x.strip())
    | beam.Map(lambda x: x.replace(' ', '_'))
)

# Good: Single map combining operations
result = data | beam.Map(lambda x: x.upper().strip().replace(' ', '_'))

// Bad: Multiple MapElements
PCollection<String> result = data
    .apply(MapElements.into(strings()).via(String::toUpperCase))
    .apply(MapElements.into(strings()).via(String::trim))
    .apply(MapElements.into(strings()).via(s -> s.replace(" ", "_")));

// Good: Single MapElements
PCollection<String> result = data.apply(
    MapElements.into(strings())
        .via(s -> s.toUpperCase().trim().replace(" ", "_"))
);

2. Use Combine instead of GroupByKey when possible

# Less efficient: GroupByKey + Map
totals = (data
    | beam.Map(lambda x: (x['key'], x['value']))
    | beam.GroupByKey()
    | beam.MapTuple(lambda k, vs: (k, sum(vs)))
)

# More efficient: CombinePerKey
totals = (data
    | beam.Map(lambda x: (x['key'], x['value']))
    | beam.CombinePerKey(sum)
)

3. Reuse expensive resources with setup/teardown

class OptimizedDoFn(DoFn):
    def setup(self):
        # Initialize expensive resources once per worker
        self.model = load_ml_model()
        self.connection_pool = create_connection_pool()

    def process(self, element):
        # Reuse resources for each element
        prediction = self.model.predict(element)
        yield prediction

    def teardown(self):
        # Clean up resources
        self.connection_pool.close()

Testing

import apache_beam as beam
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.util import assert_that, equal_to

def test_word_count():
    with TestPipeline() as p:
        input_data = ['Hello World', 'Hello Beam']
        expected_output = [('hello', 2), ('world', 1), ('beam', 1)]

        result = (p
            | beam.Create(input_data)
            | beam.FlatMap(lambda x: x.lower().split())
            | beam.Map(lambda x: (x, 1))
            | beam.CombinePerKey(sum)
        )

        assert_that(result, equal_to(expected_output))

def test_custom_dofn():
    with TestPipeline() as p:
        result = (p
            | beam.Create([1, 2, 3, 4, 5])
            | beam.ParDo(DoubleFn())
        )

        assert_that(result, equal_to([2, 4, 6, 8, 10]))

import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;

@Rule
public final transient TestPipeline pipeline = TestPipeline.create();

@Test
public void testWordCount() {
    PCollection<String> input = pipeline.apply(
        Create.of("Hello World", "Hello Beam")
    );

    PCollection<KV<String, Long>> output = input.apply(new CountWords());

    PAssert.that(output).containsInAnyOrder(
        KV.of("hello", 2L),
        KV.of("world", 1L),
        KV.of("beam", 1L)
    );

    pipeline.run();
}

Summary

In this module, you learned:

PCollections as the fundamental data abstraction in Beam
How to construct and configure pipelines
Core transforms: Map, FlatMap, Filter, GroupByKey, Combine
ParDo and DoFn for custom processing logic
DoFn lifecycle methods and advanced features
Side outputs and side inputs
Creating reusable composite transforms
Best practices for performance and testing

Key Takeaways

PCollections are immutable and distributed
Transforms create new PCollections rather than modifying existing ones
ParDo is the most flexible transform for custom logic
Use Combine instead of GroupByKey for aggregations
Leverage DoFn lifecycle methods for resource management
Write testable code using composite transforms

Next Steps

Continue to the next module on Windowing & Time to learn how to process data based on temporal characteristics and handle late-arriving data in streaming pipelines.

​Core Programming Model

​Overview

​Key Concepts

​Understanding PCollections

​PCollection Characteristics

​Creating PCollections

​PCollection Types

​Pipeline Construction

​Pipeline Creation and Configuration

​Pipeline Options

​PTransforms: Data Transformations

​Core Transform Types

​Map Transform

​FlatMap Transform

​Filter Transform

​ParDo and DoFn

​Basic DoFn

​DoFn Lifecycle Methods

​Multiple Outputs (Side Outputs)

​Side Inputs

​Composite Transforms

​Creating Composite Transforms

​Aggregation Operations

​GroupByKey

​Combine

​CoGroupByKey

​Real-World Use Cases

​Log Processing Pipeline

​ETL Pipeline

​Best Practices

​Performance Optimization

​Testing

​Summary

​Key Takeaways

​Next Steps

​Additional Resources

Core Programming Model

Overview

Key Concepts

Understanding PCollections

PCollection Characteristics

Creating PCollections

PCollection Types

Pipeline Construction

Pipeline Creation and Configuration

Pipeline Options

PTransforms: Data Transformations

Core Transform Types

Map Transform

FlatMap Transform

Filter Transform

ParDo and DoFn

Basic DoFn

DoFn Lifecycle Methods

Multiple Outputs (Side Outputs)

Side Inputs

Composite Transforms

Creating Composite Transforms

Aggregation Operations

GroupByKey

Combine

CoGroupByKey

Real-World Use Cases

Log Processing Pipeline

ETL Pipeline

Best Practices

Performance Optimization

Testing

Summary

Key Takeaways

Next Steps

Additional Resources