Trigger Types

1. Introduction

Without custom triggers, Apache Beam defaults to emitting a single output pane only when the watermark passes the end of a window. Apache Beam offers four main categories of triggers to support more complex emission patterns: Event Time, Processing Time, Element Count, and Composite.

2. Why This Concept Exists

Trigger types exist to solve specific production needs:

• Low-Latency (Speculative) Updates: In a 24-hour window, you don't want to wait until midnight to see sales results. An early trigger lets you emit hourly updates throughout the day.
• Late-Data Integration: If data arrives late due to network dropouts, late triggers let you emit revised totals.
• Flow Control: If a window suddenly accumulates millions of events (e.g. during a database dump), count triggers let you flush data early to prevent worker memory exhaustion.

3. Key Terminology

• Event-Time Trigger: Emits results based on the progress of Event Time (watermark).
• Processing-Time Trigger: Emits results based on the wall-clock system time of the worker node.
• Element-Count Trigger: Emits results based on the number of elements that have arrived in the window.
• Composite Trigger: A combination of multiple triggers utilizing boolean logic (e.g. AfterFirst, AfterAll, Repeatedly).

4. How It Works

Each trigger type evaluates a specific condition before firing a window:

• AfterWatermark: Evaluates the watermark. It fires when watermark >= window.end. In Java, it natively supports .withEarlyFirings() and .withLateFirings(). In Python, you simulate this behavior by combining triggers using AfterFirst and Repeatedly.
• AfterProcessingTime: Checks the wall-clock clock on the worker node. Fired when physical system time passes a specified delay after the first element in the window is received.
• AfterCount: Counts elements in the window. Fires as soon as the window holds $N$ elements.
• Composite (AfterFirst, AfterAll, Repeatedly):
  • AfterFirst(T1, T2): Fires when either T1 or T2 fires.
  • AfterAll(T1, T2): Fires when both T1 and T2 have fired.
  • Repeatedly(T): Re-arms the trigger T so that it fires again after emitting, allowing infinite updates.

5. Visual Diagram

6. Code Example

The following pipeline reads transactional data, segments them into hourly fixed windows, and uses a composite trigger to fire early updates every 30 seconds of system time OR every 100 elements, while still firing a final result when the watermark passes the hour:

pythonCopy
import apache_beam as beam
from apache_beam.transforms.trigger import (
    AfterWatermark,
    AfterCount,
    AfterProcessingTime,
    AfterFirst,
    Repeatedly,
    AccumulationMode
)
from apache_beam.transforms.window import FixedWindows

with beam.Pipeline() as p:
    (p
     | "ReadTransactions" >> beam.io.ReadFromPubSub(subscription="projects/my-gcp/subs/tx")
     | "MapKV" >> beam.Map(lambda x: (x["store_id"], float(x["amount"])))
     | "WindowWithCompositeTrigger" >> beam.WindowInto(
         FixedWindows(60 * 60), # 1 hour windows
         trigger=Repeatedly(
             AfterFirst(
                 AfterWatermark(),                         # Fire on-time
                 AfterCount(100),                          # Fire early if count reaches 100
                 AfterProcessingTime(30)                   # Fire early every 30 seconds
             )
         ),
         accumulation_mode=AccumulationMode.ACCUMULATING
     )
     | "SumPerStore" >> beam.CombinePerKey(sum)
     | "PrintPanes" >> beam.Map(print)
    )

7. Code Explanation

• FixedWindows(60 * 60) creates 1-hour window intervals.
• Repeatedly(...) ensures that the window does not close permanently after the first firing, letting subsequent elements continue triggering updates.
• AfterFirst(...) is a composite logical OR trigger. The window will emit an output pane (representing the current aggregate) if any of the three nested triggers evaluate to true.
• AfterWatermark() ensures that when the watermark passes the 1-hour boundary, an "On-Time" pane is fired.
• AfterCount(100) fires a speculative pane if 100 events accumulate within the 30-second window.
• AfterProcessingTime(30) fires a speculative pane every 30 seconds of wall-clock time.
• AccumulationMode.ACCUMULATING ensures that each firing contains the running total rather than just the new delta elements.

8. Real Production Example

9. Common Mistakes

• Count Triggers without Repeatedly: Using trigger=AfterCount(10) without wrapping it in Repeatedly. The runner will fire once when 10 elements arrive, and then discard all subsequent elements for that window because the trigger was not re-armed.
• High Frequency Firing Downstream: Using AfterCount(1) or extremely low processing-time delays (e.g. 100 milliseconds) on high-throughput streams. This forces the runner to write to downstream databases for every single record, destroying the database performance and nullifying the benefit of windowing.
• Omitting Accumulation Mode: Setting a trigger but forgetting to specify accumulation_mode. Apache Beam will throw an error during pipeline definition.

10. Best Practices

• Use AfterWatermark as the foundation of your trigger logic, as it aligns with event completeness.
• Limit speculative processing-time triggers to a reasonable rate (e.g. every 10 to 60 seconds) to avoid overload on downstream targets (like databases or APIs).
• Always monitor pane metadata downstream to identify if a record belongs to an early, on-time, or late emission.

11. Summary

• Triggers determine when window aggregates are emitted.
• Event-time triggers fire based on watermark progression.
• Processing-time triggers fire based on physical worker system clocks.
• Element-count triggers fire when a count threshold is hit.
• Composite triggers combine conditions using logical gates.
• Use Repeatedly to keep triggers active for multiple firings.

12. Interactive Challenges

13. Related Content

• Watermarks
• Allowed Lateness
• Accumulation Modes