Streaming, Windowing, Late Data and AI Enrichment

What it is

A streaming pipeline processes an unbounded flow of events as they arrive. Because events can arrive out of order or late, it groups them into windows by when they happened and uses watermarks and triggers to decide when a result is ready. The same pipeline can enrich each event with reference data or a machine learning prediction.

How it works

Event time, processing time and watermarks

• Event time is when the event happened, carried in the data.
• Processing time is when the pipeline sees it.
• The watermark is the pipeline's estimate that all events up to a given event time have arrived.

Business metrics almost always need event time. Counting by processing time puts an offline phone's morning events into the afternoon.

Window types

Late arriving data and triggers

Events later than the allowed lateness are dropped. If late data must never be lost, also write the raw events to a sink you can reprocess.

import json
import apache_beam as beam
from apache_beam import window
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms.trigger import (
    AccumulationMode, AfterCount, AfterProcessingTime, AfterWatermark)

with beam.Pipeline(options=PipelineOptions(streaming=True)) as p:
    (p
     | beam.io.ReadFromPubSub(
           subscription="projects/prod-data/subscriptions/clicks-sub",
           timestamp_attribute="event_ts")          # use event time
     | beam.Map(lambda msg: (json.loads(msg)["page"], 1))
     | beam.WindowInto(
           window.FixedWindows(60),
           trigger=AfterWatermark(early=AfterProcessingTime(10), late=AfterCount(1)),
           allowed_lateness=600,                     # accept events up to 10 minutes late
           accumulation_mode=AccumulationMode.ACCUMULATING)
     | beam.CombinePerKey(sum)
     | beam.Map(lambda kv: {"page": kv[0], "views": kv[1]})
     | beam.io.WriteToBigQuery("prod-data:web.page_views",
                               schema="page:STRING,views:INTEGER"))

Other window functions: window.SlidingWindows(size=300, period=60) and window.Sessions(gap_size=1800).

Processing logic

Dataflow streaming provides exactly-once processing by default and removes Pub/Sub redeliveries of the same message. A publisher that retries can still send one event twice with different message IDs, so deduplicate on a unique event ID. Pub/Sub ordering keys keep messages with the same key in order.

AI data enrichment

A worked example

A game studio counts scores per player per minute. Phones lose signal and send events up to 10 minutes late, and the leaderboard must also tag abusive chat messages.

The engineer reads Pub/Sub with the event timestamp attribute and uses fixed 1-minute windows with 10 minutes of allowed lateness and accumulating late firings. Chat messages pass through RunInference with a classification model, and flagged messages go to a moderation topic.

How the exam asks about this

• "Devices buffer events offline; reports must count each event in the minute it occurred." Event-time fixed windows using the event timestamp, with allowed lateness for late data.
• "Group clicks into visits that end after 30 minutes without activity." Session windows with a 30-minute gap.
• "Show a rolling 5-minute average refreshed every minute." Sliding windows of 5 minutes with a 1-minute period.