Dataflow Tutorial: Stream and Batch Processing on GCP

Dataflow is Google Cloud's managed service for running data pipelines, both batch and streaming. It does not have its own language. You write the pipeline with Apache Beam, an open source model, and Dataflow supplies the machines, scales them and removes them when the job ends (Google Cloud: Dataflow overview). So learning Dataflow is mostly learning Beam, and this tutorial does that with a pipeline you can run on your own computer.

The model in four words

Beam has a small vocabulary (Google Cloud: Apache Beam programming model):

• Pipeline: the whole job, a graph of steps from reading to writing.
• PCollection: a collection of elements flowing between steps. It is bounded when it has a fixed size, like a file, and unbounded when it never ends, like a stream of events.
• Transform: a step that takes collections and produces collections. Mapping, filtering, grouping and combining are transforms.
• Runner: whatever executes the pipeline. Dataflow is a runner. So is the Direct Runner, which runs on your machine.

The promise of the model is that the same pipeline code handles both kinds of collection. Batch is the case where the input ends.

A pipeline with two branches

The pipeline below reads one day of New York yellow taxi trips, 15 January 2024 (NYC TLC: trip record data), and does two things with them:

• A batch branch sums revenue per pickup zone.
• A windowed branch attaches each trip's pickup time as its event timestamp, divides time into one-hour windows and counts the trips in each. This is how a streaming pipeline aggregates.

It is run locally with the Direct Runner, not on the Dataflow service.

# An Apache Beam pipeline run locally with the Direct Runner on one day of real trip data. The same
# pipeline code can be submitted to Dataflow by changing the runner option; this is not a Dataflow run.
import glob
import shutil
import apache_beam as beam
import pandas as pd
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms import window

trips = pd.read_parquet("../../dedata/yellow_tripdata_2024-01.parquet",
                        columns=["tpep_pickup_datetime", "PULocationID", "total_amount"])
day = trips[trips.tpep_pickup_datetime.dt.strftime("%Y-%m-%d") == "2024-01-15"]
records = [{"pickup": row.tpep_pickup_datetime.timestamp(), "zone": int(row.PULocationID),
            "amount": float(row.total_amount)} for row in day.itertuples()]

shutil.rmtree("out", ignore_errors=True)
with beam.Pipeline(options=PipelineOptions(["--runner=DirectRunner"])) as pipeline:
    events = pipeline | "Read" >> beam.Create(records)

    # Batch: one result for the whole bounded collection.
    (events
     | "Valid" >> beam.Filter(lambda trip: trip["amount"] > 0)
     | "ZoneAmount" >> beam.Map(lambda trip: (trip["zone"], trip["amount"]))
     | "SumPerZone" >> beam.CombinePerKey(sum)
     | "FormatZone" >> beam.Map(lambda pair: f"{pair[1]:012.2f} zone {pair[0]}")
     | "WriteZones" >> beam.io.WriteToText("out/zones", num_shards=1, shard_name_template=""))

    # Streaming-style: the same elements, given event timestamps and grouped into one-hour windows.
    (events
     | "Stamp" >> beam.Map(lambda trip: window.TimestampedValue(trip, trip["pickup"]))
     | "Hourly" >> beam.WindowInto(window.FixedWindows(3600))
     | "One" >> beam.Map(lambda trip: 1)
     | "CountPerWindow" >> beam.CombineGlobally(sum).without_defaults()
     | "FormatHour" >> beam.Map(lambda count, w=beam.DoFn.WindowParam:
                                f"{w.start.to_utc_datetime():%H:%M} to {w.end.to_utc_datetime():%H:%M}  {count:5}")
     | "Regroup" >> beam.WindowInto(window.GlobalWindows())
     | "WriteHours" >> beam.io.WriteToText("out/hours", num_shards=1, shard_name_template=""))


def lines(prefix):
    return sorted(line.strip() for path in glob.glob(prefix + "*") for line in open(path) if line.strip())


print(f"elements read: {len(records):,}")
zones = lines("out/zones")
print(f"\nbatch branch: revenue for {len(zones)} pickup zones, top 3")
for line in zones[-1:-4:-1]:
    amount, zone = line.split(" ", 1)
    print(f"  {zone:9} {float(amount):10,.0f}")
hours = lines("out/hours")
print(f"\nwindowed branch: {len(hours)} one-hour windows, first 6")
for line in hours[:6]:
    print("  " + line)
print(f"  trips in all windows: {sum(int(line.split()[-1]) for line in hours):,}")

elements read: 77,033

batch branch: revenue for 216 pickup zones, top 3
  zone 132     453,587
  zone 138     217,892
  zone 161      79,362

windowed branch: 24 one-hour windows, first 6
  00:00 to 01:00   2237
  01:00 to 02:00   1085
  02:00 to 03:00    754
  03:00 to 04:00    443
  04:00 to 05:00    358
  05:00 to 06:00    483
  trips in all windows: 77,033

Reading the code against the output:

• The pipe operator chains transforms. Each line is one step with a name, and the name is what you will see in Dataflow's job graph.
• The key-value pair is the unit of aggregation. The batch branch maps each trip to a zone and an amount, and CombinePerKey(sum) adds the amounts for each zone. The top three are zones 132, 138 and 161: JFK Airport, LaGuardia Airport and Midtown Center.
• Windowing changes what "all the data" means. The second branch never asks for a total over everything. It asks for a count per window. The 77,033 trips fall into 24 one-hour windows, and their counts add back to 77,033.
• Both branches share one read. A pipeline is a graph, not a list. One collection can feed any number of transforms.

From your machine to Dataflow

The Direct Runner exists for development. Beam's documentation is clear that it is built for checking correctness, keeps all data in memory and is not meant for production (Apache Beam: Direct Runner).

To run the same file on Dataflow, change the options, not the code (Google Cloud: create a pipeline using Python):

python trips_pipeline.py \
  --runner DataflowRunner \
  --project analytics-prod \
  --region asia-south1 \
  --temp_location gs://example-dataflow/tmp/

Dataflow then starts worker VMs, distributes the work across them, adds or removes workers as the load changes and deletes them when the job finishes. You pay for the compute the job used.

In a real pipeline two things would differ from the example. The read would come from a source such as Cloud Storage, BigQuery or Pub/Sub instead of a list in memory, and the write would go to a table. Beam provides connectors for these.

Streaming: windows, watermarks, triggers

A stream never ends, so "group by key" has no finishing point. Streaming pipelines aggregate by window instead (Google Cloud: streaming pipelines).

The example used tumbling windows of one hour.

Windows are defined in event time, the time something happened, not the time the pipeline saw it. That raises a question the batch example never faced: when is a window finished? Events arrive out of order and late. A phone that was offline delivers an hour of events at once.

• The watermark is Dataflow's estimate of how far event time has progressed: the point before which it expects no more data. When the watermark passes the end of a window, the window's result is emitted.
• Late data is anything that arrives for a window after the watermark has passed it. By default it is dropped. You can allow a period of lateness, at the cost of holding windows open longer.
• Triggers control when results are emitted. The default is once, when the watermark passes. You can also emit early, approximate results and refine them.

This is the central trade-off of streaming, and no setting removes it: wait longer and results are more complete but later; emit sooner and they are timely but may change.

What exactly-once does and does not mean

Dataflow processes each record exactly once by default, in batch and in streaming. Within the pipeline, a retried step does not double-count (Google Cloud: exactly-once in Dataflow).

Two limits are stated in the same documentation and are worth remembering:

• Accurate is not complete. Exactly-once says that records which were processed are counted once. It cannot say that every record arrived. Late data can still be missing from a window.
• Side effects are outside the guarantee. A step that calls an external system may run more than once. Writes are exactly-once only when the sink supports it.

Templates: pipelines without code

Many jobs need no code at all. Google provides pre-built templates for common tasks, such as moving Pub/Sub messages into BigQuery, which you launch from the console with a few parameters. Your own pipelines can be packaged as templates too, so that others can run them without a development environment (Google Cloud: Dataflow templates).

Reach for a template first. Write Beam code when the logic is yours: custom parsing, enrichment against another dataset, or windowed aggregation.

Habits for a first pipeline

1. Develop on a sample with the Direct Runner, then move to Dataflow.
2. Name every transform. The names become the job graph and the first thing you read when something is slow.
3. Keep elements small and functions pure. A function that only depends on its input can be retried safely.
4. Filter early. Rows dropped in the first step cost nothing later.
5. Decide what happens to bad records. Send them to a separate output instead of failing the job.
6. Set a maximum number of workers, so a mistake cannot scale without limit.

To feed a streaming pipeline, see Pub/Sub explained. For the complete pattern, see Pub/Sub to Dataflow to BigQuery, and for the choice between this service and Spark, Dataproc vs Dataflow.

Batch and streaming pipelines are covered in the Data Engineering course.

Sources

• Dataflow overview (Google Cloud Documentation) — Google Cloud
• Programming model for Apache Beam (Google Cloud Documentation) — Google Cloud
• Streaming pipelines (Google Cloud Documentation) — Google Cloud
• Exactly-once in Dataflow (Google Cloud Documentation) — Google Cloud
• Create a Dataflow pipeline using Python (Google Cloud Documentation) — Google Cloud
• Dataflow templates (Google Cloud Documentation) — Google Cloud
• Direct Runner (Apache Beam) — Apache Beam
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission