> ## Documentation Index
> Fetch the complete documentation index at: https://resources.devweekends.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Core Programming Model

> Master PCollections and ParDo

# Core Programming Model

> [!NOTE]
>
> **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

```python Python theme={null}
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']
    )
```

```java Java theme={null}
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

```python Python theme={null}
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
```

```java Java theme={null}
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

```python Python theme={null}
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
```

```java Java theme={null}
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.

```python Python theme={null}
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)
    )
```

```java Java theme={null}
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).

```python Python theme={null}
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 [])
    )
```

```java Java theme={null}
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.

```python Python theme={null}
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)
    )
```

```java Java theme={null}
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

```python Python theme={null}
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())
    )
```

```java Java theme={null}
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.

```python Python theme={null}
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'))
    )
```

```java Java theme={null}
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.

```python Python theme={null}
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')
```

```java Java theme={null}
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.

```python Python theme={null}
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)
```

```java Java theme={null}
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

```python Python theme={null}
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)
```

```java Java theme={null}
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.

```python Python theme={null}
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))))
```

```java Java theme={null}
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.

```python Python theme={null}
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())
```

```java Java theme={null}
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.

```python Python theme={null}
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]']
```

```java Java theme={null}
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

```python Python theme={null}
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")
```

```java Java theme={null}
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

```python Python theme={null}
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**

```python Python theme={null}
# 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(' ', '_'))
```

```java Java theme={null}
// 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**

```python Python theme={null}
# 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**

```python Python theme={null}
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

```python Python theme={null}
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]))
```

```java Java theme={null}
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

1. PCollections are immutable and distributed
2. Transforms create new PCollections rather than modifying existing ones
3. ParDo is the most flexible transform for custom logic
4. Use Combine instead of GroupByKey for aggregations
5. Leverage DoFn lifecycle methods for resource management
6. 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.

***

## Additional Resources

* [Apache Beam Programming Guide](https://beam.apache.org/documentation/programming-guide/)
* [Beam Python SDK Reference](https://beam.apache.org/releases/pydoc/)
* [Beam Java SDK Reference](https://beam.apache.org/releases/javadoc/)
* [Common Beam Patterns](https://beam.apache.org/documentation/patterns/overview/)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.