Dataflow pipelines in practice

The BigQuery subscription from lesson 7 lands every web event in raw.web_events, but the operations team's request is different. They want a small table, refreshed every minute, that shows how many checkout errors each payment method produced in that minute, so they can call the payment provider the moment something breaks. That needs events grouped by time as they arrive, which is a job for a stream processor. On Thursday, October 29, Ravi introduces Dataflow.

What Dataflow is

Dataflow is Google Cloud's managed service for running data processing pipelines written with Apache Beam, an open-source programming model. Beam lets one pipeline describe a series of steps, such as read, parse, filter, group, aggregate and write, and the same code can process a fixed batch of files or an endless stream of messages. Dataflow runs the pipeline on workers it creates and removes automatically, balances the work among them, and retries failed pieces. You do not manage servers; you pay for the resources the job uses while it runs.

Templates: Dataflow without writing code

Many ingestion jobs are so common that Google provides ready-made pipelines called templates. A template is launched with parameters, from the console, the gcloud CLI or an API call, and needs no Beam code at all. Google-provided templates cover moves such as Pub/Sub to BigQuery, text files in Cloud Storage to BigQuery, a JDBC database to BigQuery, Datastream output to BigQuery, and Pub/Sub to Cloud Storage. Teams can also package their own pipelines as Flex Templates, so that analysts can run them with different parameters without touching the code. The Dataflow console also offers a job builder for assembling simple pipelines visually, and Beam pipelines can be described in YAML for straightforward cases.

Ravi's advice is to reach for the simplest tool that works. For landing web events, the BigQuery subscription already does the job, and adding Dataflow would only add cost. For the checkout summary, a custom pipeline is justified because it groups events into one-minute windows by event time, which neither a subscription nor a simple template does.

Windows, watermarks and late data

A stream never ends, so aggregations over it need boundaries. Windowing divides the stream into finite groups by event time. Beam offers three main kinds. Fixed windows are back-to-back intervals of the same length, such as each minute. Sliding windows overlap, such as a five-minute window starting every minute, which smooths a dashboard. Session windows group events from the same key that are close together and close when there is a gap, which suits shopper sessions.

Events arrive late, as lesson 7 explained, so the pipeline needs to know when a window can be considered complete. Dataflow tracks a watermark: its estimate of how far event time has progressed, meaning that events older than the watermark are not expected any more. When the watermark passes the end of a window, the window's result is emitted. Allowed lateness says how long after that the pipeline should still accept stragglers and update the result, and triggers decide when results are emitted, for example early speculative results every few seconds and a final one when the watermark passes.

# Counting checkout errors in one-minute event-time windows, with late events.
events = [  # (event_time_seconds, payment_method, arrival_time_seconds)
    (5, "CARD", 6), (20, "CARD", 21), (42, "PAYPAL", 43), (58, "CARD", 70),   # last one arrives late
    (61, "CARD", 62), (75, "CARD", 76), (119, "GIFT", 121), (30, "CARD", 140), # very late event
]
window = 60          # seconds
allowed_lateness = 30

counts, dropped = {}, []
for t, method, arrived in sorted(events, key=lambda e: e[2]):
    start = t - t % window
    window_end = start + window
    if arrived > window_end + allowed_lateness:
        dropped.append((t, method, arrived))
        continue
    key = (start, method)
    counts[key] = counts.get(key, 0) + 1

for (start, method), n in sorted(counts.items()):
    print(f"window {start:3d}-{start + window:3d}s  {method:6s} {n}")
print("dropped as too late:", dropped)

window   0- 60s  CARD   3
window   0- 60s  PAYPAL 1
window  60-120s  CARD   2
window  60-120s  GIFT   1
dropped as too late: [(30, 'CARD', 140)]

In the example, the card error that happened at 58 seconds but arrived at 70 still counts toward the first minute, because it is within the allowed lateness. The one that happened at 30 seconds but arrived at 140 is dropped from the live view. Mateo notes that nothing is lost overall: every event also lands in raw.web_events through the BigQuery subscription, so the daily reports, which run hours later, include it. The live view trades completeness for speed; the warehouse does not.

Writing the results

The pipeline writes each window's counts to curated.checkout_errors_by_minute in BigQuery using the Storage Write API. Each row carries the window start, the payment method, the count and the time it was written. When a late event updates a window, the pipeline writes a new row for that window with the updated count, so the dashboard query takes the latest row for each window. Rows the pipeline cannot parse go to a dead-letter table, raw.web_events_bad, so that a malformed message is visible rather than silently ignored.

Running and watching a job

A Dataflow job shows its pipeline as a graph in the console, with each step's throughput, the system latency (how long elements wait), the data freshness (how far the watermark lags behind real time) and the number of workers. Autoscaling adds workers when the backlog grows and removes them when it shrinks, within limits the team sets. For streaming jobs, Streaming Engine moves part of the work off the worker machines into Google's backend, which usually makes autoscaling smoother and workers smaller.

Streaming jobs also need a plan for change. When the team changes the pipeline, for example to add a new payment method, Dataflow can update a running job in place, keeping its state and its position in the subscription, as long as the new pipeline is compatible. When a job must stop, draining it stops reading new messages but finishes the work already in progress, so no windows are left half counted; cancelling stops immediately and abandons in-progress work. Before any of that, Beam pipelines can be tested on a laptop with the local direct runner and a handful of sample messages, which is far cheaper than finding a mistake in production.

What Dataflow costs

Dataflow bills for the worker resources used while a job runs, mainly vCPU time and memory, plus data processed by Dataflow Shuffle for batch jobs or by Streaming Engine for streaming jobs. A streaming job runs all the time, so even a small one costs money every hour of every day. Mateo estimates the checkout pipeline using an illustrative rate, to show the effect of keeping it small.

# Rough monthly cost of a small always-on streaming job (illustrative rates, not a quote).
hours = 24 * 30
workers = 1
vcpus_per_worker = 2
memory_gb_per_worker = 4
rate_vcpu_hour = 0.069        # dollars, illustrative
rate_gb_hour = 0.0035         # dollars, illustrative

compute = hours * workers * (vcpus_per_worker * rate_vcpu_hour + memory_gb_per_worker * rate_gb_hour)
print(f"Worker cost per month: about {compute:,.0f} dollars")
print(f"With 3 workers at peak all month: about {compute * 3:,.0f} dollars")

Worker cost per month: about 109 dollars
With 3 workers at peak all month: about 328 dollars

About 109 dollars a month for one small worker, before Streaming Engine and Pub/Sub charges, is a reasonable price for a live view the operations team says will save hours during a payment outage. The same reasoning explains why Ravi did not put the whole web-events stream through Dataflow: the BigQuery subscription does that part without an always-on job.

Dataflow, Dataproc or BigQuery?