Sessionizing a late-arriving clickstream without double-counting

Choice: Dataflow streaming, not Dataproc. The shape is unbounded, needs sub-5-minute output and event-time semantics, and the team has no Spark skills. Dataproc Serverless would force you to hand-roll watermarks and triggers for no benefit.

Do not sessionize inside the pipeline. Session windows close only after the inactivity gap (say 30 min), and with 6 h allowed lateness the window state stays open for hours. You would miss the 5-minute freshness target and hold enormous state. Instead:

• Read the Pub/Sub subscription, key by session_id, window into 1-minute event-time fixed windows.
• Trigger early (every 10 s), on-time at the watermark, and late on each late element; set allowedLateness to 6 h with accumulating panes and emit a version/sequence number so consumers take the newest pane per (window, key).
• Reassemble sessions and funnels in BigQuery over full history, where a late event is just another row.

Exactly-once. The legacy streaming insert is at-least-once, which is why the dedup job exists. Replace it with BigQueryIO using the Storage Write API with exactly-once enabled: it uses application-created streams with offset commits (Dataflow manages the offsets) instead of the default stream's at-least-once append. Dataflow already suppresses redelivered Pub/Sub messages by message ID within a running job, but that does not survive a job restart, so the sink itself must be idempotent.

If you must tolerate at-least-once: carry event_id on every row and dedup at read time with ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY ingestion_ts DESC). That works, but it reintroduces exactly the nightly batch pass you are trying to delete.

There are two independent defects; separate them before designing anything.

1. Numbers 0.3% high is a duplicate problem, not a lateness problem. Late data makes counts low. Being high means the nightly DELETE/INSERT dedup is not finishing inside its window, so duplicate rows survive to the freeze. That is a symptom of a non-idempotent write path.
2. The 02:00 cut-off cannot be exact. The day ends at 00:00 UTC; with up to 6 h of device buffering, events for that day keep arriving until 06:00 UTC. Freezing at 02:00 guarantees a 2-hour hole in the tail.

Fix the write path first. Make the sink idempotent (Storage Write API exactly-once, or a MERGE on event_id inside the streaming loop). The 4-hour dedup pass then disappears and the 0.3% error goes with it.

Then fix the calendar, not the SQL. In order of preference:

• Move the immutable freeze to 08:00 UTC (00:00 + 6 h max lateness + 2 h operating buffer). Verify the tail first: 6 h is a device maximum, so look at the P99.9 of now() - event_timestamp. If the real tail is 9 h, 08:00 is still wrong.
• If finance refuses to move, publish two artifacts: provisional at 02:00 and final at 08:00, with immutability attaching only to the final one.
• If the final artifact itself must be corrected later, correct it with an append-only adjustment ledger (signed delta rows carrying reason and run id) rather than rewriting the frozen table. The frozen table stays byte-identical; the audit trail is intact.

Finally, add a reconciliation assertion. The freeze job should compare mart counts to a count over the deduped raw archive and fail rather than publish when the delta exceeds a threshold. A blocked freeze is recoverable; a silently wrong immutable number is not.

Build on an immutable raw zone. Archive Pub/Sub to gs://raw/events/dt=YYYY-MM-DD/hour=HH/ via a dedicated subscription or a pipeline branch. At 12,000 msg/s x 1.5 KB you write ~1.5 TB/day, so rotate hourly, move to Nearline at 30 days and Coldline at 90. The archive is the only thing that survives a bad deploy.

Backfill as a separate job, never by editing the live one.

• A batch Dataflow job reads the affected date range from GCS and writes to staging.events_backfill, partitioned by DATE(event_date) and clustered identically to the target.
• Publish with a single MERGE into production restricted to the affected partitions, matching on event_id. One statement means readers see the pre- or post-backfill state, never a half-written partition.
• Never use WRITE_APPEND for a correction; it double-counts.

Bound the blast radius and the bill. A full two-year pass is roughly 1.1 PB of raw data; at on-demand $6.25/TB that is about $7,000 and many hours per pass. So: reprocess only the partition range whose logic or source data changed; select only the columns the transform needs (columnar reads plus clustering make this several times cheaper than SELECT *); run the backfill with FlexRS to use Spot capacity; and require a WHERE event_date BETWEEN predicate on the staging table so a missing filter fails fast instead of scanning everything.

Make it repeatable. Parameterize the job by date range and output-table suffix so a failed run can be restarted from scratch with no cleanup, and swap in only at the final MERGE. Idempotency comes from event_id, not from careful bookkeeping.