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

# Triggers & Watermarks

> Control result materialization

# Triggers & Watermarks

> [!NOTE]
>
> **Module Duration**: 3-4 hours
> **Focus**: Advanced timing control and late data handling
> **Prerequisites**: Core Programming Model, Windowing & Time

## Overview

Triggers determine when results are materialized for a window. They provide fine-grained control over when computations complete and emit results, enabling you to balance latency, completeness, and cost.

### Key Concepts

* **Trigger**: Determines when to emit aggregated results
* **Watermark**: System's notion of event time progress
* **Early Firing**: Emit speculative results before window closes
* **On-Time Firing**: Emit results when watermark passes window end
* **Late Firing**: Handle data arriving after watermark
* **Accumulation Mode**: How to combine multiple firings

***

## Understanding Watermarks

Watermarks are the foundation of Beam's event-time processing. They represent a heuristic about event time completeness.

### Watermark Semantics

**Definition**: A watermark with value T indicates that no more elements with timestamps less than T should arrive.

**Properties**:

* Monotonically increasing
* Per-source tracking
* Heuristic-based (not a guarantee)

### Watermark Behavior

```python Python theme={null}
import apache_beam as beam
from apache_beam import window

# Default watermark behavior
with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromPubSub(topic='events')
        | beam.Map(parse_event)
        # Watermark automatically tracked
        | beam.WindowInto(window.FixedWindows(60))
        | beam.CombinePerKey(sum)
    )

# Watermark advances as:
# 1. Source reads events and assigns timestamps
# 2. Beam tracks the minimum timestamp of pending data
# 3. Watermark = min(pending timestamps) - some buffer
# 4. When watermark > window.end, window closes
```

```java Java theme={null}
// Watermark is automatically managed by Beam
PCollection<Event> events = pipeline.apply(
    PubsubIO.readStrings()
        .fromTopic("projects/my-project/topics/events")
).apply(ParDo.of(new ParseEventFn()));

// Watermark advances based on:
// 1. Event timestamps
// 2. Source characteristics
// 3. Processing rate
// 4. Runner implementation
```

### Watermark Lag

Watermark lag represents how far behind the watermark is from real time.

```python Python theme={null}
# Monitoring watermark lag
import apache_beam as beam
from apache_beam import DoFn

class MonitorWatermarkLagFn(DoFn):
    def process(self, element, timestamp=DoFn.TimestampParam):
        import time
        current_time = time.time()
        event_time = timestamp.micros / 1_000_000
        lag = current_time - event_time

        yield {
            'element': element,
            'event_time': event_time,
            'processing_time': current_time,
            'watermark_lag_seconds': lag
        }
```

***

## Default Trigger

The default trigger fires when the watermark passes the end of the window.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.Create([
            (1, 'a'), (59, 'b'), (61, 'c'), (120, 'd')
        ])
        | beam.Map(lambda x: window.TimestampedValue(x[1], x[0]))
        | beam.WindowInto(window.FixedWindows(60))
        # Default trigger: fires when watermark passes window end
        | beam.CombineGlobally(lambda xs: list(xs)).without_defaults()
    )

    # Window [0, 60): fires when watermark >= 60
    #   Contains: 'a', 'b'
    # Window [60, 120): fires when watermark >= 120
    #   Contains: 'c'
    # Window [120, 180): fires when watermark >= 180
    #   Contains: 'd'
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterWatermark;

PCollection<String> windowed = data
    .apply(Window.<String>into(FixedWindows.of(Duration.standardSeconds(60)))
        // Default trigger (explicit)
        .triggering(AfterWatermark.pastEndOfWindow())
        .withAllowedLateness(Duration.ZERO)
        .discardingFiredPanes());
```

***

## Trigger Types

Beam provides several trigger types that can be combined to create sophisticated timing strategies.

### AfterWatermark

Fires when the watermark passes the end of the window.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromPubSub(topic='events')
        | beam.Map(parse_event)
        | beam.WindowInto(
            window.FixedWindows(60),
            trigger=trigger.AfterWatermark(),
            accumulation_mode=trigger.AccumulationMode.DISCARDING
        )
        | beam.CombinePerKey(sum)
    )
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
import org.apache.beam.sdk.transforms.windowing.Window;

PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(1)))
        .triggering(AfterWatermark.pastEndOfWindow())
        .discardingFiredPanes()
);
```

### AfterProcessingTime

Fires after a certain amount of processing time has elapsed.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromPubSub(topic='events')
        | beam.Map(parse_event)
        | beam.WindowInto(
            window.FixedWindows(300),  # 5-minute windows
            trigger=trigger.AfterProcessingTime(60),  # Fire every minute
            accumulation_mode=trigger.AccumulationMode.ACCUMULATING
        )
        | beam.CombinePerKey(sum)
    )

    # Fires every 60 seconds of processing time
    # regardless of event time or watermark
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;

PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterProcessingTime.pastFirstElementInPane()
                .plusDelayOf(Duration.standardSeconds(60))
        )
        .accumulatingFiredPanes()
);
```

### AfterCount

Fires after a certain number of elements arrive in a pane.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromPubSub(topic='events')
        | beam.Map(parse_event)
        | beam.WindowInto(
            window.FixedWindows(60),
            trigger=trigger.AfterCount(100),  # Fire after 100 elements
            accumulation_mode=trigger.AccumulationMode.DISCARDING
        )
        | beam.CombinePerKey(sum)
    )
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterPane;

PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(1)))
        .triggering(AfterPane.elementCountAtLeast(100))
        .discardingFiredPanes()
);
```

### Repeatedly

Fires a trigger repeatedly.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromPubSub(topic='events')
        | beam.Map(parse_event)
        | beam.WindowInto(
            window.FixedWindows(300),
            trigger=trigger.Repeatedly(
                trigger.AfterProcessingTime(30)  # Fire every 30 seconds
            ),
            accumulation_mode=trigger.AccumulationMode.ACCUMULATING
        )
        | beam.CombinePerKey(sum)
    )
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.Repeatedly;

PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            Repeatedly.forever(
                AfterProcessingTime.pastFirstElementInPane()
                    .plusDelayOf(Duration.standardSeconds(30))
            )
        )
        .accumulatingFiredPanes()
);
```

***

## Early and Late Firings

Combine early, on-time, and late firings for flexible result emission.

### Early Firing Pattern

Emit speculative results before the watermark passes the window end.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

def early_firing_pipeline():
    with beam.Pipeline() as pipeline:
        data = (pipeline
            | beam.io.ReadFromKafka(
                consumer_config={'bootstrap.servers': 'localhost:9092'},
                topics=['events']
            )
            | beam.Map(parse_event)
            | beam.WindowInto(
                window.FixedWindows(300),  # 5-minute windows
                trigger=trigger.AfterWatermark(
                    early=trigger.AfterProcessingTime(60)  # Early: every minute
                ),
                accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
                allowed_lateness=0
            )
            | beam.CombinePerKey(sum)
        )

        data | beam.Map(print)

# Timeline for a 5-minute window [0:00, 5:00):
# 1:00 - Early firing (partial results)
# 2:00 - Early firing (updated partial results)
# 3:00 - Early firing (updated partial results)
# 4:00 - Early firing (updated partial results)
# 5:00 - On-time firing (watermark passed, final result)
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;

PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterWatermark.pastEndOfWindow()
                .withEarlyFirings(
                    AfterProcessingTime.pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1))
                )
        )
        .withAllowedLateness(Duration.ZERO)
        .accumulatingFiredPanes()
);

// Fires:
// - Every 1 minute (early)
// - When watermark passes window end (on-time)
```

### Late Firing Pattern

Handle data arriving after the watermark has passed.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

def late_firing_pipeline():
    with beam.Pipeline() as pipeline:
        data = (pipeline
            | beam.io.ReadFromPubSub(topic='events')
            | beam.Map(parse_event)
            | beam.WindowInto(
                window.FixedWindows(300),
                trigger=trigger.AfterWatermark(
                    late=trigger.AfterProcessingTime(60)  # Late: every minute
                ),
                accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
                allowed_lateness=600  # Allow 10 minutes of lateness
            )
            | beam.CombinePerKey(sum)
        )

        data | beam.Map(print)

# Timeline for a 5-minute window [0:00, 5:00):
# 5:00 - On-time firing (watermark passed)
# 6:00 - Late firing (late data arrived)
# 7:00 - Late firing (more late data)
# ...
# 15:00 - Window closes (allowed lateness expired)
```

```java Java theme={null}
PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterWatermark.pastEndOfWindow()
                .withLateFirings(
                    AfterProcessingTime.pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1))
                )
        )
        .withAllowedLateness(Duration.standardMinutes(10))
        .accumulatingFiredPanes()
);
```

### Complete Early/Late Pattern

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

def complete_early_late_pipeline():
    with beam.Pipeline() as pipeline:
        data = (pipeline
            | beam.io.ReadFromKafka(...)
            | beam.Map(parse_event)
            | beam.WindowInto(
                window.FixedWindows(300),  # 5-minute windows
                trigger=trigger.AfterWatermark(
                    early=trigger.AfterProcessingTime(60),  # Early: every minute
                    late=trigger.AfterCount(100)  # Late: every 100 elements
                ),
                accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
                allowed_lateness=600  # 10 minutes
            )
            | beam.CombinePerKey(sum)
        )

        # Add metadata to track firing type
        tagged_data = data | beam.ParDo(AddFiringMetadataFn())

        tagged_data | beam.io.WriteToBigQuery(
            table='analytics.windowed_results'
        )

from apache_beam import DoFn, PaneInfo

class AddFiringMetadataFn(DoFn):
    def process(self, element, pane_info=DoFn.PaneInfoParam):
        key, value = element

        firing_type = 'unknown'
        if pane_info.is_first and not pane_info.is_last:
            firing_type = 'early'
        elif not pane_info.is_first and not pane_info.is_last:
            firing_type = 'late'
        elif pane_info.is_last:
            firing_type = 'final'
        elif pane_info.is_first and pane_info.is_last:
            firing_type = 'on_time'

        yield {
            'key': key,
            'value': value,
            'firing_type': firing_type,
            'is_first': pane_info.is_first,
            'is_last': pane_info.is_last,
            'index': pane_info.index
        }
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.PaneInfo;

PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterWatermark.pastEndOfWindow()
                .withEarlyFirings(
                    AfterProcessingTime.pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1))
                )
                .withLateFirings(AfterPane.elementCountAtLeast(100))
        )
        .withAllowedLateness(Duration.standardMinutes(10))
        .accumulatingFiredPanes()
);

static class AddFiringMetadataFn extends DoFn<KV<String, Integer>, Result> {
    @ProcessElement
    public void processElement(@Element KV<String, Integer> element,
                               @PaneInfo PaneInfo paneInfo,
                               OutputReceiver<Result> out) {
        String firingType;
        if (paneInfo.isFirst() && !paneInfo.isLast()) {
            firingType = "early";
        } else if (!paneInfo.isFirst() && !paneInfo.isLast()) {
            firingType = "late";
        } else if (paneInfo.isLast()) {
            firingType = "final";
        } else {
            firingType = "on_time";
        }

        Result result = new Result(
            element.getKey(),
            element.getValue(),
            firingType,
            paneInfo.getIndex()
        );

        out.output(result);
    }
}
```

***

## Accumulation Modes

Accumulation modes determine how multiple firings of the same window are combined.

### Discarding Mode

Each firing contains only new data since the last firing.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.Create([
            (0, 1), (30, 2), (60, 3), (90, 4)
        ])
        | beam.Map(lambda x: window.TimestampedValue(x[1], x[0]))
        | beam.WindowInto(
            window.FixedWindows(120),
            trigger=trigger.Repeatedly(
                trigger.AfterProcessingTime(30)
            ),
            accumulation_mode=trigger.AccumulationMode.DISCARDING
        )
        | beam.CombineGlobally(sum).without_defaults()
    )

    # Firings for window [0, 120):
    # 30s: sum([1]) = 1 (first element)
    # 60s: sum([2]) = 2 (only new element)
    # 90s: sum([3]) = 3 (only new element)
    # 120s: sum([4]) = 4 (only new element)
```

```java Java theme={null}
PCollection<Integer> windowed = data.apply(
    Window.<Integer>into(FixedWindows.of(Duration.standardSeconds(120)))
        .triggering(
            Repeatedly.forever(
                AfterProcessingTime.pastFirstElementInPane()
                    .plusDelayOf(Duration.standardSeconds(30))
            )
        )
        .discardingFiredPanes()  // Discarding mode
);
```

### Accumulating Mode

Each firing contains all data seen so far for the window.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.Create([
            (0, 1), (30, 2), (60, 3), (90, 4)
        ])
        | beam.Map(lambda x: window.TimestampedValue(x[1], x[0]))
        | beam.WindowInto(
            window.FixedWindows(120),
            trigger=trigger.Repeatedly(
                trigger.AfterProcessingTime(30)
            ),
            accumulation_mode=trigger.AccumulationMode.ACCUMULATING
        )
        | beam.CombineGlobally(sum).without_defaults()
    )

    # Firings for window [0, 120):
    # 30s: sum([1]) = 1
    # 60s: sum([1, 2]) = 3 (cumulative)
    # 90s: sum([1, 2, 3]) = 6 (cumulative)
    # 120s: sum([1, 2, 3, 4]) = 10 (cumulative)
```

```java Java theme={null}
PCollection<Integer> windowed = data.apply(
    Window.<Integer>into(FixedWindows.of(Duration.standardSeconds(120)))
        .triggering(
            Repeatedly.forever(
                AfterProcessingTime.pastFirstElementInPane()
                    .plusDelayOf(Duration.standardSeconds(30))
            )
        )
        .accumulatingFiredPanes()  // Accumulating mode
);
```

### Accumulating and Retracting

Similar to accumulating, but also emits retractions for previous firings.

```python Python theme={null}
# Note: Accumulating and retracting is primarily available in Java SDK
# Python SDK has limited support for retractions
```

```java Java theme={null}
PCollection<Integer> windowed = data.apply(
    Window.<Integer>into(FixedWindows.of(Duration.standardSeconds(120)))
        .triggering(
            Repeatedly.forever(
                AfterProcessingTime.pastFirstElementInPane()
                    .plusDelayOf(Duration.standardSeconds(30))
            )
        )
        .accumulatingAndRetractingFiredPanes()
);

// Firings for window [0, 120):
// 30s: +1 (emit 1)
// 60s: -1, +3 (retract 1, emit 3)
// 90s: -3, +6 (retract 3, emit 6)
// 120s: -6, +10 (retract 6, emit 10)
```

***

## Composite Triggers

Combine multiple triggers with logical operators.

### AfterEach (Sequential)

Fire triggers in sequence.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromPubSub(topic='events')
        | beam.Map(parse_event)
        | beam.WindowInto(
            window.FixedWindows(300),
            trigger=trigger.AfterEach(
                trigger.AfterCount(1000),  # First: fire after 1000 elements
                trigger.Repeatedly(
                    trigger.AfterProcessingTime(60)  # Then: every minute
                )
            ),
            accumulation_mode=trigger.AccumulationMode.ACCUMULATING
        )
        | beam.CombinePerKey(sum)
    )
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterEach;

PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterEach.inOrder(
                AfterPane.elementCountAtLeast(1000),
                Repeatedly.forever(
                    AfterProcessingTime.pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1))
                )
            )
        )
        .accumulatingFiredPanes()
);
```

### AfterAll (Conjunction)

Fire when all triggers have fired.

```python Python theme={null}
# AfterAll is less commonly used and has limited Python support
# Typically use AfterWatermark with early/late firings instead
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterAll;

// Fire when BOTH conditions are met
PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterAll.of(
                AfterPane.elementCountAtLeast(100),
                AfterProcessingTime.pastFirstElementInPane()
                    .plusDelayOf(Duration.standardMinutes(2))
            )
        )
        .discardingFiredPanes()
);
```

### AfterAny (Disjunction)

Fire when any trigger fires.

```python Python theme={null}
# Python SDK: Use OrFinally for similar behavior
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromPubSub(topic='events')
        | beam.Map(parse_event)
        | beam.WindowInto(
            window.FixedWindows(300),
            trigger=trigger.OrFinally(
                trigger.AfterCount(1000),  # Fire after 1000 elements
                trigger.AfterProcessingTime(120)  # OR after 2 minutes
            ),
            accumulation_mode=trigger.AccumulationMode.DISCARDING
        )
        | beam.CombinePerKey(sum)
    )
```

```java Java theme={null}
import org.apache.beam.sdk.transforms.windowing.AfterFirst;

// Fire when EITHER condition is met
PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterFirst.of(
                AfterPane.elementCountAtLeast(1000),
                AfterProcessingTime.pastFirstElementInPane()
                    .plusDelayOf(Duration.standardMinutes(2))
            )
        )
        .discardingFiredPanes()
);
```

***

## Allowed Lateness

Control how long to keep window state after the watermark passes.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

with beam.Pipeline() as pipeline:
    data = (pipeline
        | beam.io.ReadFromKafka(...)
        | beam.Map(parse_event)
        | beam.WindowInto(
            window.FixedWindows(300),
            trigger=trigger.AfterWatermark(
                late=trigger.AfterCount(1)  # Fire on each late element
            ),
            accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
            allowed_lateness=600  # Keep state for 10 minutes after watermark
        )
        | beam.CombinePerKey(sum)
    )

# Window [0:00, 5:00):
# 5:00 - Watermark passes, on-time firing
# 5:01 - Late element arrives, late firing
# ...
# 14:59 - Late element arrives, late firing
# 15:00 - allowed_lateness expires, drop state
# 15:01 - Any late elements after this are dropped
```

```java Java theme={null}
PCollection<KV<String, Integer>> windowed = data.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .triggering(
            AfterWatermark.pastEndOfWindow()
                .withLateFirings(AfterPane.elementCountAtLeast(1))
        )
        .withAllowedLateness(Duration.standardMinutes(10))
        .accumulatingFiredPanes()
);

// Data dropped if it arrives more than 10 minutes after watermark
```

***

## Real-World Use Cases

### Low-Latency Dashboard

Provide quick updates with early firings, refine with late data.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

def low_latency_dashboard():
    with beam.Pipeline() as pipeline:
        events = (pipeline
            | "Read Events" >> beam.io.ReadFromKafka(
                consumer_config={'bootstrap.servers': 'localhost:9092'},
                topics=['user-events']
            )
            | "Parse" >> beam.Map(parse_event)
        )

        # Real-time metrics with early/late firings
        metrics = (events
            | "1-Min Windows" >> beam.WindowInto(
                window.FixedWindows(60),
                trigger=trigger.AfterWatermark(
                    early=trigger.AfterProcessingTime(10),  # Update every 10s
                    late=trigger.AfterCount(50)  # Update every 50 late elements
                ),
                accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
                allowed_lateness=300  # 5 minutes
            )
            | "Count Events" >> beam.combiners.Count.PerKey()
            | "Add Metadata" >> beam.ParDo(AddDashboardMetadataFn())
        )

        # Write to real-time dashboard
        metrics | "Write to Dashboard" >> beam.io.WriteToPubSub(
            topic='dashboard-metrics'
        )

class AddDashboardMetadataFn(beam.DoFn):
    def process(self, element, pane_info=beam.DoFn.PaneInfoParam,
                window=beam.DoFn.WindowParam):
        key, count = element

        # Determine update type
        update_type = 'final'
        if pane_info.is_first and not pane_info.is_last:
            update_type = 'preliminary'
        elif not pane_info.is_first and not pane_info.is_last:
            update_type = 'updated'

        yield {
            'metric_key': key,
            'count': count,
            'update_type': update_type,
            'window_start': window.start.to_utc_datetime().isoformat(),
            'window_end': window.end.to_utc_datetime().isoformat(),
            'firing_index': pane_info.index,
            'is_final': pane_info.is_last
        }
```

### Session Analysis with Late Data

Handle user sessions with realistic late-arrival handling.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

def session_analysis_with_late_data():
    with beam.Pipeline() as pipeline:
        clicks = (pipeline
            | "Read Clicks" >> beam.io.ReadFromPubSub(topic='clickstream')
            | "Parse" >> beam.Map(parse_click)
        )

        # Session windows with late data handling
        sessions = (clicks
            | "Session Windows" >> beam.WindowInto(
                window.Sessions(gap_size=600),  # 10-minute gap
                trigger=trigger.AfterWatermark(
                    early=trigger.AfterProcessingTime(120),  # Updates every 2 min
                    late=trigger.AfterCount(1)  # Fire on each late click
                ),
                accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
                allowed_lateness=1800  # 30 minutes
            )
            | "Key by User" >> beam.Map(lambda x: (x['user_id'], x))
            | "Group Sessions" >> beam.GroupByKey()
            | "Analyze" >> beam.ParDo(SessionAnalysisFn())
        )

        # Separate early/on-time/late results
        early_sessions = sessions | "Filter Early" >> beam.Filter(
            lambda x: x['firing_type'] == 'early'
        )
        final_sessions = sessions | "Filter Final" >> beam.Filter(
            lambda x: x['firing_type'] in ['on_time', 'late']
        )

        # Early results for real-time monitoring
        early_sessions | "Write Early" >> beam.io.WriteToPubSub(
            topic='session-updates'
        )

        # Final results for analytics
        final_sessions | "Write Final" >> beam.io.WriteToBigQuery(
            table='analytics.user_sessions'
        )

class SessionAnalysisFn(beam.DoFn):
    def process(self, element, pane_info=beam.DoFn.PaneInfoParam):
        user_id, clicks = element
        clicks_list = list(clicks)

        firing_type = 'on_time'
        if pane_info.is_first and not pane_info.is_last:
            firing_type = 'early'
        elif not pane_info.is_first:
            firing_type = 'late'

        yield {
            'user_id': user_id,
            'click_count': len(clicks_list),
            'unique_pages': len(set(c['page'] for c in clicks_list)),
            'session_duration': (
                max(c['timestamp'] for c in clicks_list) -
                min(c['timestamp'] for c in clicks_list)
            ),
            'firing_type': firing_type,
            'firing_index': pane_info.index
        }
```

### Trading Analytics

High-frequency updates with precise control.

```python Python theme={null}
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms import trigger

def trading_analytics():
    with beam.Pipeline() as pipeline:
        trades = (pipeline
            | "Read Trades" >> beam.io.ReadFromKafka(
                consumer_config={'bootstrap.servers': 'localhost:9092'},
                topics=['market-trades']
            )
            | "Parse" >> beam.Map(parse_trade)
        )

        # 1-second windows with sub-second updates
        second_stats = (trades
            | "1-Sec Windows" >> beam.WindowInto(
                window.FixedWindows(1),  # 1-second windows
                trigger=trigger.AfterWatermark(
                    early=trigger.AfterProcessingTime(0.1)  # Every 100ms
                ),
                accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
                allowed_lateness=5  # 5 seconds
            )
            | "Key by Symbol" >> beam.Map(lambda x: (x['symbol'], x['price']))
            | "Calculate VWAP" >> beam.CombinePerKey(VWAPCombineFn())
        )

        # 1-minute windows for aggregated stats
        minute_stats = (trades
            | "1-Min Windows" >> beam.WindowInto(
                window.FixedWindows(60),
                trigger=trigger.AfterWatermark(
                    early=trigger.AfterProcessingTime(5)  # Every 5 seconds
                ),
                accumulation_mode=trigger.AccumulationMode.ACCUMULATING
            )
            | "Key by Symbol (Min)" >> beam.Map(
                lambda x: (x['symbol'], x)
            )
            | "Aggregate Minute" >> beam.CombinePerKey(
                MinuteStatsCombineFn()
            )
        )

        # Real-time updates
        second_stats | "Publish Second Stats" >> beam.io.WriteToPubSub(
            topic='market-stats-1s'
        )

        # Persistent storage
        minute_stats | "Write Minute Stats" >> beam.io.WriteToBigQuery(
            table='market.minute_stats'
        )

from apache_beam.transforms import CombineFn

class VWAPCombineFn(CombineFn):
    """Volume-Weighted Average Price."""

    def create_accumulator(self):
        return {'total_value': 0.0, 'total_volume': 0}

    def add_input(self, accumulator, element):
        price = element
        # Assuming unit volume for simplicity
        accumulator['total_value'] += price
        accumulator['total_volume'] += 1
        return accumulator

    def merge_accumulators(self, accumulators):
        merged = self.create_accumulator()
        for acc in accumulators:
            merged['total_value'] += acc['total_value']
            merged['total_volume'] += acc['total_volume']
        return merged

    def extract_output(self, accumulator):
        if accumulator['total_volume'] == 0:
            return 0.0
        return accumulator['total_value'] / accumulator['total_volume']

class MinuteStatsCombineFn(CombineFn):
    def create_accumulator(self):
        return {
            'trades': [],
            'volume': 0,
            'total_value': 0.0
        }

    def add_input(self, accumulator, trade):
        accumulator['trades'].append(trade['price'])
        accumulator['volume'] += trade.get('volume', 1)
        accumulator['total_value'] += trade['price'] * trade.get('volume', 1)
        return accumulator

    def merge_accumulators(self, accumulators):
        merged = self.create_accumulator()
        for acc in accumulators:
            merged['trades'].extend(acc['trades'])
            merged['volume'] += acc['volume']
            merged['total_value'] += acc['total_value']
        return merged

    def extract_output(self, accumulator):
        if not accumulator['trades']:
            return None

        prices = sorted(accumulator['trades'])
        return {
            'open': accumulator['trades'][0],
            'high': max(prices),
            'low': min(prices),
            'close': accumulator['trades'][-1],
            'volume': accumulator['volume'],
            'vwap': accumulator['total_value'] / accumulator['volume']
        }
```

***

## Best Practices

### Choosing Trigger Strategies

**Use Default Trigger When**:

* Latency is not critical
* You can wait for watermark
* Data completeness is more important than speed

**Use Early Firings When**:

* Low latency is required
* Users expect real-time updates
* Approximate results are acceptable

**Use Late Firings When**:

* Data frequently arrives late
* You need to update results
* Completeness is critical

### Balancing Latency and Completeness

```python Python theme={null}
# Low latency (more frequent updates, less complete)
low_latency = beam.WindowInto(
    window.FixedWindows(60),
    trigger=trigger.AfterWatermark(
        early=trigger.AfterProcessingTime(5)  # Very frequent
    ),
    accumulation_mode=trigger.AccumulationMode.ACCUMULATING
)

# Balanced (moderate latency, good completeness)
balanced = beam.WindowInto(
    window.FixedWindows(300),
    trigger=trigger.AfterWatermark(
        early=trigger.AfterProcessingTime(60),
        late=trigger.AfterCount(100)
    ),
    accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
    allowed_lateness=600
)

# High completeness (wait for watermark, handle late data)
high_completeness = beam.WindowInto(
    window.FixedWindows(3600),
    trigger=trigger.AfterWatermark(
        late=trigger.AfterCount(1)
    ),
    accumulation_mode=trigger.AccumulationMode.ACCUMULATING,
    allowed_lateness=3600
)
```

### Resource Management

```python Python theme={null}
# Careful with allowed lateness - impacts state retention
# Too long: High memory usage
too_long = beam.WindowInto(
    window.FixedWindows(60),
    allowed_lateness=86400  # 24 hours - expensive!
)

# Too short: May lose important late data
too_short = beam.WindowInto(
    window.FixedWindows(60),
    allowed_lateness=0  # No late data handling
)

# Reasonable: Balance completeness and resources
reasonable = beam.WindowInto(
    window.FixedWindows(60),
    allowed_lateness=300  # 5 minutes
)
```

***

## Summary

In this module, you learned:

* Watermark semantics and behavior
* Trigger types: AfterWatermark, AfterProcessingTime, AfterCount
* Early, on-time, and late firings
* Accumulation modes: discarding vs accumulating
* Composite triggers for complex timing logic
* Allowed lateness for state management
* Real-world trigger strategies

### Key Takeaways

1. Triggers control when to emit results for a window
2. Watermarks track event time progress
3. Early firings reduce latency, late firings improve completeness
4. Accumulation mode determines how multiple firings combine
5. Allowed lateness balances completeness and resource usage
6. Choose trigger strategy based on latency/completeness requirements

### Next Steps

Continue to the next module on **State & Timers** to learn how to maintain per-key state and implement complex stateful processing patterns.

***

## Additional Resources

* [Beam Triggers Documentation](https://beam.apache.org/documentation/programming-guide/#triggers)
* [Watermarks in Apache Beam](https://beam.apache.org/documentation/programming-guide/#watermarks-and-late-data)
* [The World Beyond Batch](https://www.oreilly.com/radar/the-world-beyond-batch-streaming-102/)
* [Trigger Examples](https://beam.apache.org/documentation/programming-guide/#composite-triggers)


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