Processing Time

1. Introduction

Unlike Event Time (which is determined by the data itself), Processing Time is determined entirely by the system clock of the runner's computing resources.

2. Why This Concept Exists

While Event Time is required for business-accurate aggregations, Processing Time is critical for operations, system monitoring, and latency-sensitive alerting.

Processing Time is highly valuable when:

• Low Latency is Critical: You need to emit data immediately without waiting for late-arriving elements or watermark progression.
• Pipeline Health Monitoring: You want to measure operational metrics like CPU usage, disk IO, or the current processing rate of your workers.
• Rate Limiting / Throttling: You need to throttle API calls or database writes to a maximum number of requests per physical wall-clock second.
• Ingestion-based grouping: You don't care when the event actually happened, only that you bucket events as they arrive to write them to storage quickly.

3. Key Terminology

• Processing Time: The time of execution according to the physical worker's clock.
• Wall-Clock Time: The real-world date and time (standardized to UTC) of a computer's system clock.
• Clock Drift: Small discrepancies between the clocks of different worker machines in a distributed cluster.
• Ingestion Time: The timestamp assigned when an event enters the system queue (e.g. Pub/Sub), which sits between Event Time and Processing Time.

4. How It Works

1. Arrival: An event arrives at a worker machine.
2. Stamping: When the event enters a processing step, the runner inspects the local system clock (wall-clock time).
3. Assignment: If processing-time windowing is configured, the element is assigned to a time window based on this system clock reading.
4. Instant Execution: Because physical time moves forward constantly and uniformly, processing-time windows close and fire immediately as soon as the wall-clock time passes the window end. No watermarks are needed to advance time.

5. Visual Diagram

6. Code Example

The following pipeline overrides incoming event timestamps with the current wall-clock processing time, segments them into 1-minute fixed windows, and writes them to a console log:

pythonCopy
import time
import apache_beam as beam
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms.window import TimestampedValue

class AssignProcessingTimestamp(beam.DoFn):
    def process(self, element):
        # Read the current worker wall-clock time
        current_wall_clock = time.time()
        # Overwrite the element's metadata timestamp with processing time
        yield TimestampedValue(element, current_wall_clock)

with beam.Pipeline() as p:
    (p
     | "ReadRealTimeEvents" >> beam.io.ReadFromPubSub(subscription="projects/my-gcp/subs/live-feed")
     | "StampProcessingTime" >> beam.ParDo(AssignProcessingTimestamp())
     | "WindowProcessing" >> beam.WindowInto(FixedWindows(60)) # 1 minute
     | "CountEvents" >> beam.CombineGlobally(sum).without_defaults()
     | "LogRates" >> beam.Map(print))

7. Code Explanation

• time.time() returns the epoch float value of the current machine.
• TimestampedValue(element, current_wall_clock) overrides the element's timestamp, resetting it to the current wall-clock time.
• beam.WindowInto(FixedWindows(60)) creates 1-minute buckets based on this new processing timestamp.
• The results are processed and emitted immediately after each physical minute passes.

8. Real Production Example

9. Common Mistakes

• Using Processing Time for Backfills: Rerunning a processing-time pipeline on historical logs. Because the system clock is used, logs from three months ago will be bucketed into today's current wall-clock window, rendering historical aggregations meaningless.
• Assuming Perfect Synchronization: Assuming that all workers have the exact same time. If NTP is disabled, clock drift between nodes can cause elements to be placed in different windows on different workers.
• Using Processing Time for Financial Ledger Audits: Utilizing processing-time windowing to compute financial reporting. Since late network packets fall into later processing windows, the reports will not match the physical transaction history.

10. Best Practices

• Never use Processing Time for business intelligence or historical reporting; use it only for operations, alerting, and rate-limiting.
• Keep NTP (Network Time Protocol) active on all cluster machines to minimize clock drift.
• Combine processing-time triggers with event-time windowing to get the best of both worlds (early speculative results with eventual event-time accuracy).

11. Summary

• Processing Time is based on the system clock of the worker node.
• It offers the lowest possible processing latency.
• It is non-deterministic and depends entirely on pipeline execution speed.
• It is unsuitable for historical replays or backfills.
• Useful for rate-limiting, infrastructure alerting, and low-latency metrics.

12. Interactive Challenges

13. Related Content

• Why Windowing?
• Event Time
• Watermarks