> ## 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

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

## 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

<CodeGroup>
  ```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")
  );
  ```
</CodeGroup>

### 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

<CodeGroup>
  ```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();
  ```
</CodeGroup>

### Pipeline Options

<CodeGroup>
  ```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);
  ```
</CodeGroup>

***

## 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.

<CodeGroup>
  ```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)
  );
  ```
</CodeGroup>

### FlatMap Transform

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

<CodeGroup>
  ```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()))
  );
  ```
</CodeGroup>

### Filter Transform

Keep only elements matching a predicate.

<CodeGroup>
  ```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)
  );
  ```
</CodeGroup>

***

## 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

<CodeGroup>
  ```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())
  );
  ```
</CodeGroup>

### DoFn Lifecycle Methods

DoFn provides lifecycle methods for setup and teardown operations.

<CodeGroup>
  ```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";
      }
  }
  ```
</CodeGroup>

### Multiple Outputs (Side Outputs)

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

<CodeGroup>
  ```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);
  ```
</CodeGroup>

### Side Inputs

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

<CodeGroup>
  ```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)
  );
  ```
</CodeGroup>

***

## Composite Transforms

Composite transforms encapsulate multiple transforms into reusable components.

### Creating Composite Transforms

<CodeGroup>
  ```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));
  ```
</CodeGroup>

***

## Aggregation Operations

Beam provides several built-in transforms for aggregating data.

### GroupByKey

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

<CodeGroup>
  ```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())))
  );
  ```
</CodeGroup>

### Combine

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

<CodeGroup>
  ```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()));
  ```
</CodeGroup>

### CoGroupByKey

Joins multiple PCollections with the same key type.

<CodeGroup>
  ```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;
          })
  );
  ```
</CodeGroup>

***

## Real-World Use Cases

### Log Processing Pipeline

<CodeGroup>
  ```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()));
          }
      }
  }
  ```
</CodeGroup>

### ETL Pipeline

<CodeGroup>
  ```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')
  ```
</CodeGroup>

***

## Best Practices

### Performance Optimization

**1. Minimize data serialization**

<CodeGroup>
  ```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(" ", "_"))
  );
  ```
</CodeGroup>

**2. Use Combine instead of GroupByKey when possible**

<CodeGroup>
  ```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)
  )
  ```
</CodeGroup>

**3. Reuse expensive resources with setup/teardown**

<CodeGroup>
  ```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()
  ```
</CodeGroup>

### Testing

<CodeGroup>
  ```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();
  }
  ```
</CodeGroup>

***

## 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/)