Clickstream Funnels With Skewed Device Clocks

Two consumers with opposite needs, so build two outputs, not one compromise.

Dashboard path (latency wins): session windows with a 30-minute gap, withAllowedLateness(60 min) - set above the observed 40-minute skew so the watermark does not declare a slow device's events late prematurely. Fire early panes every minute in accumulating mode so a 5-minute refresh shows something. Label these numbers as provisional in the UI; they will revise.

Finance path (correctness wins): read nothing from the streaming aggregate. Compute from the deduplicated raw table only after the watermark has passed the period plus allowed lateness (i.e. settle D-1 on D+1).

Clock skew handling: event time, never processing time. Events stamped more than N minutes in the future are quarantined to a side output rather than allowed to hold or advance the watermark.

90-day replay: Pub/Sub retention is measured in days, not months, so raw events must also land in GCS/BigQuery at ingest. That archive is the replay source; the streaming aggregate is derived and disposable.

The judgement to state is: do not try to make one aggregate serve both. Sessionization + allowed lateness gives correct-once-settled data, not correct-at-5-minutes data.

Non-determinism usually enters through three doors, and all three are fixable:

1. State TTL vs. dedup window. If dedup or session state expires on wall-clock/processing time, a re-run sees a different set of retained keys and drops a different subset of duplicates. Key dedup on event_id within a window derived from event time, or dedup against a durable seen_events table.
2. Reading derived data as input. If the new run reads last week's aggregate instead of raw events, it inherits whatever the old buggy run emitted. Always rebuild derived tables from the immutable archive.
3. Non-deterministic transforms. CURRENT_TIMESTAMP(), RAND(), unordered window ordering, and unordered ARRAY_AGG all change between runs. Fix the ordering key and pass the run's logical date as a parameter, not the clock.

Make the output idempotent rather than merely deterministic: write each output day with WRITE_TRUNCATE on that day's partition (or MERGE on the natural key), so a re-run of day X replaces day X exactly. Stamp the transform version into the output rows so a mixed-version table is detectable.

Then the acceptance test is concrete: re-run day X twice and diff - zero rows, zero metric drift.

Work from the largest multiplier down, and measure before changing anything.

Suspects, in order:

• Sink batching. A 5-minute requirement implemented as 1-minute triggers produces 5x the BigQuery load jobs, each with overhead, plus more shuffle churn. Set triggering frequency to the refresh the product actually shows.
• Shuffle and streaming runner settings. Dataflow Shuffle and Streaming Engine move state and shuffle off the workers, which typically allows far fewer, smaller workers for a stateful sessionization job. Verify by comparing worker CPU and shuffle-backed hotspot metrics, not by assumption.
• Over-provisioned or badly keyed workers. Hot keys (one huge referrer domain) pin a single worker while the rest idle. Add a combiner/pre-aggregation before the sink and salt pathological keys.
• Streaming inserts vs. bulk load. For the raw archive, staging to GCS and running load jobs is materially cheaper per row than streaming inserts.

The structural option worth costing out: keep Pub/Sub → GCS as a cheap durable capture, and run sessionization as a 5-minute micro-batch over the raw table (Dataflow batch or Dataproc Serverless). You trade up to 5 minutes of latency you are already tolerating for a batch-priced compute bill, and you get replay for free because the source is the archive.

State the tradeoff explicitly: the streaming graph buys sub-minute freshness the product does not need, and that is where the 3x lives.