Pub/Sub to Dataflow to BigQuery: A Streaming Pattern

Events arrive in Pub/Sub, a Dataflow pipeline cleans and reshapes them, and the results land in BigQuery within seconds. This is the reference streaming design on Google Cloud, and it appears in most architecture diagrams. It is also often built when it is not needed. This guide starts with that question, then works through what the middle stage has to do, using a pipeline you can run locally.

Do you need the middle?

Pub/Sub can write to BigQuery by itself. A BigQuery subscription delivers each message to a table with no pipeline running anywhere (Google Cloud: BigQuery subscriptions). It is cheaper and has nothing to operate.

Add Dataflow when the messages need work that a subscription cannot do:

Google's guidance draws the same line: Dataflow is recommended where more complex transformation is required, especially windowing or aggregating across messages. If your requirement is "get the events into a table", start without it. See loading data into BigQuery.

What the pipeline has to do

When Dataflow is justified, a production pipeline has five jobs:

1. Read from a subscription.
2. Parse and validate, sending what cannot be parsed to a dead-letter destination instead of failing.
3. De-duplicate, because Pub/Sub delivers at least once.
4. Window and aggregate, if the output is a summary and not raw events.
5. Write to BigQuery.

The script below implements stages 2 to 4 and runs them locally. A scripted test stream stands in for the subscription, and a text file for the table. It is Apache Beam on the Direct Runner, not a run on Pub/Sub, Dataflow or BigQuery. The stream contains six messages, one of them malformed and one delivered twice, followed by a seventh that arrives late, after the watermark has passed its window.

# The Pub/Sub to Dataflow to BigQuery pattern as an Apache Beam pipeline, run locally with the Direct
# Runner on a scripted test stream. Nothing here touches Pub/Sub, Dataflow or BigQuery: the stream
# stands in for a subscription and a text file for the table. The watermark is advanced by hand.
import json
import os
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.testing.test_stream import TestStream
from apache_beam.transforms import trigger, window
from apache_beam.transforms.window import TimestampedValue

MESSAGES = [                                              # (arrival order, payload); ts is event time
    '{"id": "e1", "store": "hyd", "amount": 100, "ts": 5}',
    '{"id": "e2", "store": "hyd", "amount": 50, "ts": 20}',
    '{"id": "e2", "store": "hyd", "amount": 50, "ts": 20}',   # the same message delivered twice
    'not json at all',                                        # a malformed message
    '{"id": "e3", "store": "blr", "amount": 70, "ts": 40}',
    '{"id": "e4", "store": "hyd", "amount": 30, "ts": 70}',
]
LATE = '{"id": "e5", "store": "hyd", "amount": 20, "ts": 50}'  # arrives after its window has closed


class Parse(beam.DoFn):
    def process(self, payload):
        try:
            event = json.loads(payload)
            yield TimestampedValue(event, event["ts"])            # event time comes from the message
        except (ValueError, KeyError):
            yield beam.pvalue.TaggedOutput("dead_letter", payload)


def write(path, line):
    with open(path, "a") as table:                               # stands in for the BigQuery write
        table.write(line + "\n")


def run(allowed_lateness, path):
    if os.path.exists(path):
        os.remove(path)
    options = PipelineOptions(["--runner=DirectRunner"])
    options.view_as(StandardOptions).streaming = True
    stream = (TestStream().advance_watermark_to(0).add_elements(MESSAGES)
              .advance_watermark_to(125)                         # both one-minute windows are now past
              .add_elements([LATE]).advance_watermark_to_infinity())
    with beam.Pipeline(options=options) as pipeline:
        parsed = pipeline | stream | "Parse" >> beam.ParDo(Parse()).with_outputs("dead_letter", main="events")
        parsed.dead_letter | "DeadLetter" >> beam.Map(lambda payload: write(path, f"dead letter: {payload}"))
        (parsed.events
         | "Window" >> beam.WindowInto(window.FixedWindows(60), allowed_lateness=allowed_lateness,
                                       trigger=trigger.AfterWatermark(late=trigger.AfterCount(1)),
                                       accumulation_mode=trigger.AccumulationMode.ACCUMULATING)
         | "KeyById" >> beam.Map(lambda event: (event["id"], event))
         | "GroupById" >> beam.GroupByKey()
         | "KeepOne" >> beam.Map(lambda pair: (list(pair[1])[0]["store"], list(pair[1])[0]["amount"]))
         | "SumPerStore" >> beam.CombinePerKey(sum)
         | "Row" >> beam.Map(lambda pair, w=beam.DoFn.WindowParam, pane=beam.DoFn.PaneInfoParam: write(
             path, f"window {int(w.start):3}-{int(w.end):3}  store {pair[0]}  total {pair[1]:3}  "
                   f"{'late update' if pane.timing == 2 else 'on time'}")))
    return sorted(open(path).read().splitlines())


for lateness in (0, 600):
    print(f"allowed lateness {lateness} seconds")
    for line in run(lateness, f"table_{lateness}.txt"):
        print("  " + line)

allowed lateness 0 seconds
  dead letter: not json at all
  window   0- 60  store blr  total  70  on time
  window   0- 60  store hyd  total 150  on time
  window  60-120  store hyd  total  30  on time
allowed lateness 600 seconds
  dead letter: not json at all
  window   0- 60  store blr  total  70  on time
  window   0- 60  store hyd  total 150  on time
  window   0- 60  store hyd  total 170  late update
  window  60-120  store hyd  total  30  on time

Reading the result

The malformed message does not stop the pipeline. It goes to the dead-letter output and the rest carry on. Without that branch the failing message would be retried. In streaming mode Dataflow retries a failing bundle indefinitely, which can stall the pipeline permanently (Google Cloud: pipeline lifecycle).

The duplicate is counted once. Event e2 was delivered twice. The total for the first window is 150, not 200, because the pipeline groups by event ID before summing.

The late event depends on one setting. Event e5 belongs to the first window and arrives after that window has closed.

• With allowed lateness of 0, it is dropped. The table says 150 and will always say 150.
• With allowed lateness of 600 seconds, the window is kept open for stragglers. A second row appears, marked as a late update, with the corrected total of 170.

Neither is wrong. The first gives a table whose rows never change. The second gives a more complete answer, in exchange for results that can be revised, which every consumer of the table must be prepared for. This is the decision that defines a streaming pipeline, and it should be made by whoever owns the numbers, not left to a default.

The real connectors

In production the test stream and the file are replaced by the Pub/Sub and BigQuery connectors:

import json
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions(streaming=True)
with beam.Pipeline(options=options) as pipeline:
    (pipeline
     | "Read" >> beam.io.ReadFromPubSub(
           subscription="projects/analytics-prod/subscriptions/orders-to-dataflow",
           id_label="event_id",
           timestamp_attribute="event_time")
     | "Parse" >> beam.Map(json.loads)
     | "Write" >> beam.io.WriteToBigQuery(
           "analytics-prod:sales.order_events",
           method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API,
           create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
           write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

Three choices in those few lines matter.

De-duplication by your own ID. Dataflow removes redelivered messages by Pub/Sub message ID. That does not catch a publisher that sent the same event twice, because the two messages have different IDs. Naming an attribute that carries your event ID extends de-duplication to that case, for messages published within 10 minutes of each other (Google Cloud: read from Pub/Sub to Dataflow).

Event time from the message. By default the connector uses the time Pub/Sub received the message. Naming a timestamp attribute makes windows follow when the event happened, which is what a business report means by "per hour".

The write method. The connector offers several (Google Cloud: write from Dataflow to BigQuery):

The same documentation makes two operational points worth repeating. Give each pipeline its own subscription; two pipelines sharing one receive arbitrary halves of the data. And do not enable Pub/Sub's own exactly-once delivery on a subscription read by Dataflow, which handles duplicates itself.

Or use the template

If the transformation is no more than mapping JSON fields to columns, Google's Pub/Sub to BigQuery template does it without code. It reads JSON messages, writes to an existing table and sends failures to a dead-letter table (Google Cloud: Pub/Sub to BigQuery template). The template's own page suggests considering a BigQuery subscription first, which is a fair summary of this whole article.

Exactly-once, at-least-once and cost

Dataflow's streaming default is exactly-once processing. It can be switched to at-least-once, which lowers cost and latency by skipping de-duplication (Google Cloud: exactly-once in Dataflow).

Choose by what a duplicate costs. For payments or stock movements, keep exactly-once. For clickstream that is aggregated approximately anyway, at-least-once may be the sensible economy. Remember what neither mode promises: completeness. A late event can be missing from a window in either mode (Google Cloud: streaming pipelines).

A streaming pipeline also runs all day, whether or not events arrive, which a batch job does not. Before building one, ask how fresh the data must really be. A load every five minutes is far simpler and satisfies many "real-time" requirements.

Operating it

• Watch the backlog. The age of the oldest unacknowledged message is the first sign that the pipeline is falling behind.
• Watch the dead-letter destination. A sudden rise usually means the publisher changed its message format.
• Keep raw events. Write the unmodified messages to a table or bucket as well. When the transformation logic turns out to be wrong, you can reprocess.
• Plan updates. Changing a running streaming pipeline is harder than redeploying a batch job. Practise it before you need it.
• Partition the destination table by event date, so that late updates and reprocessing touch only the affected days. See partitioning and clustering.

For each service on its own, see Pub/Sub explained and the Dataflow tutorial.

Streaming pipelines are covered in the Data Engineering course.

Sources

• BigQuery subscriptions (Google Cloud Documentation) — Google Cloud
• Read from Pub/Sub to Dataflow (Google Cloud Documentation) — Google Cloud
• Write from Dataflow to BigQuery (Google Cloud Documentation) — Google Cloud
• Pub/Sub to BigQuery template (Google Cloud Documentation) — Google Cloud
• Streaming pipelines (Google Cloud Documentation) — Google Cloud
• Exactly-once in Dataflow (Google Cloud Documentation) — Google Cloud
• Pipeline lifecycle (Google Cloud Documentation) — Google Cloud