How to Stream Data from Pub/Sub into BigQuery

Pub/Sub to BigQuery is one of the most common streaming patterns on GCP — events published to a topic land in a BigQuery table within seconds. This guide walks through the exact steps to build it.

Architecture

Producer (app/service)
    → Pub/Sub Topic
        → Dataflow Streaming Job (Pub/Sub to BigQuery template)
            → BigQuery Table

Dataflow handles exactly-once delivery, schema validation, and dead-letter routing for malformed messages. You do not write any pipeline code — GCP provides a managed template.

Prerequisites

• ▸GCP project with billing enabled
• ▸gcloud CLI installed and authenticated
• ▸APIs enabled: Pub/Sub, Dataflow, BigQuery, Cloud Storage

Enable APIs in one command:

gcloud services enable pubsub.googleapis.com \
    dataflow.googleapis.com \
    bigquery.googleapis.com \
    storage.googleapis.com

Step 1: Create the Pub/Sub Topic and Subscription

PROJECT_ID=your-project-id
REGION=asia-south1
TOPIC=events-topic
SUBSCRIPTION=events-sub

# Create topic
gcloud pubsub topics create $TOPIC

# Create subscription (Dataflow will use this)
gcloud pubsub subscriptions create $SUBSCRIPTION \
    --topic=$TOPIC \
    --ack-deadline=60

Step 2: Create the BigQuery Dataset and Table

Your messages must be JSON. Define a schema that matches them.

Example message:

{
  "user_id": "u_12345",
  "event": "page_view",
  "url": "/courses/python",
  "timestamp": "2026-04-06T10:30:00Z"
}

Create the dataset and table:

DATASET=analytics
TABLE=events

# Create dataset
bq mk --location=asia-south1 $DATASET

# Create table with schema
bq mk --table $DATASET.$TABLE \
    user_id:STRING,\
    event:STRING,\
    url:STRING,\
    timestamp:TIMESTAMP

Or use a JSON schema file for complex tables:

bq mk --table $DATASET.$TABLE schema.json

Step 3: Create a Cloud Storage Bucket for Dataflow Staging

Dataflow needs a GCS bucket for temporary files and staging artifacts.

BUCKET=gs://${PROJECT_ID}-dataflow-staging

gcloud storage buckets create $BUCKET \
    --location=$REGION \
    --uniform-bucket-level-access

Step 4: Launch the Dataflow Streaming Job

GCP provides a pre-built template: Pub/Sub Subscription to BigQuery. No pipeline code needed.

JOB_NAME=pubsub-to-bq-events

gcloud dataflow jobs run $JOB_NAME \
    --gcs-location gs://dataflow-templates-${REGION}/latest/PubSub_Subscription_to_BigQuery \
    --region $REGION \
    --staging-location ${BUCKET}/staging \
    --temp-location ${BUCKET}/temp \
    --parameters \
        inputSubscription=projects/${PROJECT_ID}/subscriptions/${SUBSCRIPTION},\
        outputTableSpec=${PROJECT_ID}:${DATASET}.${TABLE},\
        outputDeadletterTable=${PROJECT_ID}:${DATASET}.events_dead_letter

The outputDeadletterTable captures messages that fail schema validation — essential for debugging.

The job will appear in the Dataflow console within 2-3 minutes. Streaming jobs run continuously until you cancel them.

Step 5: Publish a Test Message

gcloud pubsub topics publish $TOPIC \
    --message='{"user_id":"u_001","event":"page_view","url":"/courses/python","timestamp":"2026-04-06T10:30:00Z"}'

Within 10-30 seconds, the row appears in BigQuery:

SELECT * FROM `your-project.analytics.events`
ORDER BY timestamp DESC
LIMIT 10;

Step 6: Publish from Your Application

Python producer example:

from google.cloud import pubsub_v1
import json
from datetime import datetime, timezone

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("your-project-id", "events-topic")

def publish_event(user_id: str, event: str, url: str):
    message = {
        "user_id": user_id,
        "event": event,
        "url": url,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }
    data = json.dumps(message).encode("utf-8")
    future = publisher.publish(topic_path, data)
    return future.result()  # waits for publish confirmation

publish_event("u_12345", "course_start", "/courses/python")

Monitoring

Watch your pipeline in the Dataflow console — it shows throughput, lag, and error rates visually.

Set up a BigQuery query to check ingestion lag:

SELECT
    TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), MAX(timestamp), SECOND) as lag_seconds,
    COUNT(*) as events_last_minute
FROM `your-project.analytics.events`
WHERE timestamp > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 MINUTE)

Monitor dead-letter messages — if this table grows, your message format has changed:

SELECT * FROM `your-project.analytics.events_dead_letter` LIMIT 20;

Cost Estimate

For a modest pipeline (1000 events/minute):

• ▸Pub/Sub: ~$0.04/month (first 10GB free)
• ▸Dataflow: ~$30-50/month (n1-standard-2 worker, always running)
• ▸BigQuery storage: ~$0.02/GB/month

The Dataflow streaming job is the main cost. For very low-volume use cases, consider the BigQuery Pub/Sub Direct subscription (currently in preview) — it routes messages directly without Dataflow.

What to Do Next

• ▸Add schema evolution — use BigQuery's schema auto-detection or manage versions explicitly
• ▸Set table expiration on the dead-letter table so it does not grow unbounded
• ▸Add partitioning on the timestamp column to keep query costs low as data grows
• ▸Set up a Pub/Sub dead-letter topic for messages that cannot be delivered to the subscription