{"id":9397,"date":"2026-09-21T09:33:24","date_gmt":"2026-09-21T09:33:24","guid":{"rendered":"https:\/\/prwatech.in\/blog\/?p=9397"},"modified":"2026-09-22T15:10:42","modified_gmt":"2026-09-22T15:10:42","slug":"pubsub-stream-processing-with-dataflow","status":"publish","type":"post","link":"https:\/\/prwatech.in\/blog\/google-cloud-platform\/dataflow\/pubsub-stream-processing-with-dataflow\/","title":{"rendered":"Pubsub stream processing with dataflow"},"content":{"rendered":"\r\n

Streaming Pub\/Sub Messages to Storage With Dataflow<\/h1>\r\n

This builds a streaming pipeline, one that runs continuously rather than once on a fixed dataset, reading messages published to a Pub\/Sub topic and writing them into Cloud Storage in timed batches. It is a different shape of problem from the batch ETL pipeline covered in the ETL Processing With Dataflow and BigQuery guide in this series, which runs once against a file that already exists. Here, the pipeline keeps running and keeps writing new output as new messages arrive.<\/p>\r\n

This walks through publishing test messages on a schedule, running the streaming pipeline against them, watching it write output to a bucket, and shutting everything down again.<\/p>\r\n

Step by Step Process of Streaming Pub\/Sub Messages<\/h2>\r\n

Step 1: Activate Cloud Shell<\/h2>\r\n

If a permission error comes up later, grant the Owner role to the default compute service account before continuing.<\/p>\r\n

Step 2: Set Your Variables<\/h2>\r\n

BUCKET_NAME=your-bucket-name
PROJECT_ID=$(gcloud config get-value project)
TOPIC_ID=your-topic-id
REGION=your-dataflow-region<\/p>\r\n

Choose a Dataflow region close to wherever you are actually running these commands from.<\/h3>\r\n\r\n\r\n\r\n
\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 3: Create a Cloud Scheduler Job<\/h3>\r\n

This publishes a test message to your Pub\/Sub topic once a minute:<\/p>\r\n

gcloud scheduler jobs create pubsub publisher-job \\
–schedule=”* * * * *” \\
–topic=$TOPIC_ID –message-body=”Hello!”<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 3: Start the Job<\/h3>\r\n

gcloud scheduler jobs run publisher-job<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 5: Clone the Sample Repository<\/h3>\r\n

git clone https:\/\/github.com\/GoogleCloudPlatform\/python-docs-samples.git<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 6: Move Into the Sample Directory<\/h3>\r\n

cd python-docs-samples\/pubsub\/streaming-analytics<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 7: Install the Requirements<\/h3>\r\n

pip3 install -r requirements.txt<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

If a required Apache Beam version cannot be found, install the current version instead of pinning an old one. Open requirements.txt with nano requirements.txt and either remove a hard version number entirely or update it to a current release, rather than setting it back to 2.24.0.<\/p>\r\n

Step 8: Edit the Script<\/h3>\r\n

Open the file:<\/p>\r\n

nano PubSubToGCS.py<\/p>\r\n

Replace its contents with the following, which is the corrected version of the script the live page’s copy had lost its formatting from:<\/p>\r\n

import argparse
from datetime import datetime
import logging
import random

from apache_beam import DoFn, GroupByKey, io, ParDo, Pipeline, PTransform, WindowInto, WithKeys
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms.window import FixedWindows


class GroupMessagesByFixedWindows(PTransform):
\u00a0\u00a0\u00a0\u00a0def __init__(self, window_size, num_shards=5):
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0self.window_size = int(window_size * 60)
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0self.num_shards = num_shards

\u00a0\u00a0\u00a0\u00a0def expand(self, pcoll):
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0return (
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0pcoll
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0| “Window into fixed intervals” >> WindowInto(FixedWindows(self.window_size))
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0| “Add timestamp to windowed elements” >> ParDo(AddTimestamp())
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0| “Add key” >> WithKeys(lambda _: random.randint(0, self.num_shards – 1))
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0| “Group by key” >> GroupByKey()
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0)


class AddTimestamp(DoFn):
\u00a0\u00a0\u00a0\u00a0def process(self, element, publish_time=DoFn.TimestampParam):
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0yield (
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0element.decode(“utf-8”),
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0datetime.utcfromtimestamp(float(publish_time)).strftime(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0“%Y-%m-%d %H:%M:%S.%f”
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0),
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0)


class WriteToGCS(DoFn):
\u00a0\u00a0\u00a0\u00a0def __init__(self, output_path):
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0self.output_path = output_path

\u00a0\u00a0\u00a0\u00a0def process(self, key_value, window=DoFn.WindowParam):
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0ts_format = “%H:%M”
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0window_start = window.start.to_utc_datetime().strftime(ts_format)
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0window_end = window.end.to_utc_datetime().strftime(ts_format)
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0shard_id, batch = key_value
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0filename = “-“.join([self.output_path, window_start, window_end, str(shard_id)])

\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0with io.gcsio.GcsIO().open(filename=filename, mode=”w”) as f:
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0for message_body, publish_time in batch:
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0f.write(f”{message_body},{publish_time}”.encode(“utf-8”))


def run(input_topic, output_path, window_size=1.0, num_shards=5, pipeline_args=None):
\u00a0\u00a0\u00a0\u00a0pipeline_options = PipelineOptions(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0pipeline_args, streaming=True, save_main_session=True
\u00a0\u00a0\u00a0\u00a0)

\u00a0\u00a0\u00a0\u00a0with Pipeline(options=pipeline_options) as pipeline:
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0pipeline
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0| “Read from Pub\/Sub” >> io.ReadFromPubSub(topic=input_topic)
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0| “Window into” >> GroupMessagesByFixedWindows(window_size, num_shards)
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0| “Write to GCS” >> ParDo(WriteToGCS(output_path))
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0)


if __name__ == “__main__”:
\u00a0\u00a0\u00a0\u00a0logging.getLogger().setLevel(logging.INFO)

\u00a0\u00a0\u00a0\u00a0parser = argparse.ArgumentParser()
\u00a0\u00a0\u00a0\u00a0parser.add_argument(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0“–input_topic”,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0help=”The Cloud Pub\/Sub topic to read from, in the format projects\/PROJECT_ID\/topics\/TOPIC_ID.”,
\u00a0\u00a0\u00a0\u00a0)
\u00a0\u00a0\u00a0\u00a0parser.add_argument(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0“–window_size”,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0type=float,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0default=1.0,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0help=”Output file’s window size in minutes.”,
\u00a0\u00a0\u00a0\u00a0)
\u00a0\u00a0\u00a0\u00a0parser.add_argument(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0“–output_path”,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0help=”Path of the output GCS file including the prefix.”,
\u00a0\u00a0\u00a0\u00a0)
\u00a0\u00a0\u00a0\u00a0parser.add_argument(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0“–num_shards”,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0type=int,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0default=5,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0help=”Number of shards to use when writing windowed elements to GCS.”,
\u00a0\u00a0\u00a0\u00a0)
\u00a0\u00a0\u00a0\u00a0known_args, pipeline_args = parser.parse_known_args()

\u00a0\u00a0\u00a0\u00a0run(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0known_args.input_topic,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0known_args.output_path,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0known_args.window_size,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0known_args.num_shards,
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0pipeline_args,<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n
\r\n
\r\n
\r\n
\r\n
\r\n
\r\n

Save and exit with control x, then y, then enter.<\/p>\r\n<\/div>\r\n<\/div>\r\n<\/div>\r\n<\/div>\r\n<\/div>\r\n<\/div>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 9: Run the Streaming Pipeline<\/h3>\r\n

python3 PubSubToGCS.py \\
–project=$PROJECT_ID \\
–region=$REGION \\
–input_topic=projects\/$PROJECT_ID\/topics\/$TOPIC_ID \\
–output_path=gs:\/\/$BUCKET_NAME\/samples\/output \\
–runner=DataflowRunner \\
–window_size=2 \\
–num_shards=2 \\
–temp_location=gs:\/\/$BUCKET_NAME\/temp<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 10: Watch the Job in Dataflow<\/h3>\r\n

Open the menu, then Dataflow, to see the job’s progress.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 11: Check the Output<\/h3>\r\n

After a short wait, check the bucket for a samples folder with output inside it:<\/p>\r\n

gsutil ls gs:\/\/${BUCKET_NAME}\/samples\/<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 12: Clean Up<\/h3>\r\n

Delete the scheduler job so it stops publishing test messages:<\/p>\r\n

gcloud scheduler jobs delete publisher-job<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n

Streaming Compared to Batch Pipelines<\/h2>\r\n

A batch pipeline, such as the one in the ETL Processing With Dataflow and BigQuery guide, runs once against data that already exists, finishes, and stops. A streaming pipeline, like this one, starts and keeps running indefinitely, processing new data as it arrives in fixed time windows, here every two minutes as set by window_size. Streaming suits ongoing sources such as a live event feed, while batch suits a fixed file or table you already have in hand.<\/p>\r\n

Common Mistakes to Avoid<\/h2>\r\n