Timers & Scheduling

1. Introduction

2. Why This Concept Exists

Stateful processing allows you to store data, but you also need a way to act based on the passage of time. For example:

• Timeouts: Alerting if an IoT device fails to send a heartbeat within 5 minutes.
• Garbage Collection: Clearing old state cells to free memory if a user session goes inactive.
• Window overrides: Forcing results to emit at exact intervals. Timers provide the mechanism to schedule and trigger these time-based operations.

3. Key Terminology

• Event-Time Timers: Fire based on watermark progress. Fired only when the stream's watermark passes the registered timestamp.
• Processing-Time Timers: Fire based on real-world wall-clock time on the worker machine.
• Timer Spec: The declaration of the timer slot (TimerSpec).
• On-Timer Callback: The method annotated with @on_timer that executes when the timer fires.

4. How It Works

1. Declare: Define a TimerSpec at the class level of your stateful DoFn.
2. Schedule: Inside process(), set the timer to a specific timestamp using timer.set(timestamp).
3. Execute: When the target time arrives, Beam executes the callback method associated with that timer.

5. Visual Diagram

6. Code Example

Setting an event-time timer to fire 5 minutes after an element's timestamp:

pythonCopy
import apache_beam as beam
from apache_beam.state import TimerSpec, ValueStateSpec
from apache_beam.coders import VarIntCoder
from apache_beam.transforms.trigger import TimeDomain

class TimeoutAlertDoFn(beam.DoFn):
    # 1. Declare state and timer specs
    LAST_SEEN_STATE = ValueStateSpec("last_seen", VarIntCoder())
    ALERT_TIMER = TimerSpec("alert_timer", TimeDomain.EVENT_TIME)

    def process(self, element, 
                timestamp=beam.DoFn.TimestampParam,
                last_seen=beam.DoFn.StateParam(LAST_SEEN_STATE),
                timer=beam.DoFn.TimerParam(ALERT_TIMER)):
        key, val = element
        
        # Save timestamp
        last_seen.write(int(timestamp))
        
        # 2. Schedule timer to fire 5 minutes (300 seconds) in event time
        timer.set(int(timestamp) + 300)
        yield element

    # 3. Define callback method triggered when timer fires
    @on_timer(ALERT_TIMER)
    def trigger_alert(self, key = beam.DoFn.KeyParam):
        # Fire alert payload
        yield (key, "ALERT: No events seen in last 5 minutes!")

7. Code Explanation

• ALERT_TIMER = TimerSpec("alert_timer", TimeDomain.EVENT_TIME) creates an Event-Time timer cell.
• timer.set(...) schedules the callback execution at 5 minutes past the element's timestamp.
• @on_timer(ALERT_TIMER) binds the trigger_alert function to the timer, receiving the corresponding key parameter.

8. Real Production Example

9. Common Mistakes

• Setting Timers without keys: Like stateful processing, timers require key-value inputs. Applying them to non-keyed data throws a validation crash.
• Neglecting timezone offsets: When using processing-time timers, be careful with system local time zones on worker VMs. Always stick to UTC.

10. Best Practices

• Always clear state cells associated with a key inside the timer callback to prevent leaks.
• Use Event-Time timers for backfills and deterministic testing, and Processing-Time timers solely for wall-clock notifications.

11. Summary

• Timers schedule callbacks based on time domains (Event/Processing).
• Configured using @on_timer(TimerSpec).
• Crucial for session timeouts, heartbeats, and state memory cleanup.

12. Interactive Challenges

13. Related Content

• Stateful Processing
• Watermarks