Getting Started with GCP Dataflow for Batch Pipelines: A Practical Guide

When you are building data pipelines on GCP, you will eventually hit a point where Cloud Functions or simple scheduled scripts don’t cut it anymore. Maybe your CSV files grew from a few MBs to a few GBs, or you need to join multiple datasets and the old approach takes 40 minutes and keeps running out of memory. That is where Dataflow comes in.

In this article, we will go through what Dataflow is, set up a simple batch pipeline using the Apache Beam Python SDK, run it locally, and then deploy it to GCP. By the end, you should have a decent idea of where Dataflow fits and what to watch out for when you move beyond a demo.

What is Dataflow?

Dataflow is GCP’s managed service for running Apache Beam pipelines. If you haven’t used Beam before, it is an open-source unified programming model for both batch and stream processing. You write your pipeline once using the Beam SDK, and you can run it on different runners — the DirectRunner for local testing, the Dataflow runner for GCP, or even Flink or Spark runners if you want.

The big sell is that you don’t manage any infrastructure. You give Dataflow your code, it spins up worker VMs, autoscales them, handles failures, and tears them down when the job is done.

When Should You Use Dataflow?

Not every pipeline needs Dataflow. Here is a rough guide based on what I have seen in practice:

Scenario	Use This	Why
Small CSV/JSON file, simple transform, runs in under 5 minutes	Cloud Functions / Cloud Run	Dataflow would be overkill — worker startup alone takes a couple of minutes
Moderate data (a few GBs), joins, aggregations, runs on a schedule	Dataflow (batch)	Handles the scale, retries on failure, good for daily/hourly jobs
Streaming data, low-latency processing	Dataflow (streaming)	Exactly what it is built for — Pub/Sub to BigQuery with windowing
Complex DAG with many dependencies	Cloud Composer + Dataflow	Composer orchestrates, Dataflow does the heavy lifting on individual steps
You already have Spark expertise and existing Spark jobs	Dataproc	No need to rewrite working code in Beam unless there is a clear benefit

The key thing is that Dataflow jobs have a startup overhead. Spinning up workers takes around 2-4 minutes, so for tiny jobs that run in 30 seconds, you are paying more for startup than for actual work. For anything that genuinely needs parallel processing, though, it is worth it.

Prerequisites

Before we write any code, make sure you have:

1. A GCP project with billing enabled
2. The Dataflow API enabled (go to APIs & Services → Enable APIs → search for “Dataflow”)
3. A GCS bucket for staging files and temporary storage
4. The gcloud CLI installed and authenticated
5. Python 3.8+ and a virtual environment

Also, the Dataflow service needs a service account with the right permissions. By default, it uses the Compute Engine default service account. For a real project, create a dedicated service account with at least the Dataflow Worker and Storage Object Admin roles.

Setting Up the Python Environment

Let us set up a clean environment and install the Beam SDK with the GCP extras:

mkdir dataflow-batch-demo && cd dataflow-batch-demo
python3 -m venv venv
source venv/bin/activate
pip install apache-beam[gcp]

The [gcp] extra pulls in the dependencies needed to talk to GCS, BigQuery, and the Dataflow runner. Without it, you can only use the DirectRunner locally.

Writing a Simple Batch Pipeline

Let us build something realistic. Imagine you have CSV files in a GCS bucket containing sales transactions, and you want to aggregate revenue by product category and write the result to BigQuery. Nothing fancy, but it covers reading, transforming, grouping, and writing — which is 90% of what batch pipelines do.

Here is the pipeline code:

import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
from apache_beam.io.textio import ReadFromText
from apache_beam.io.gcp.bigquery import WriteToBigQuery

class ParseTransaction(beam.DoFn):
    def process(self, line):
        # CSV format: transaction_id,product_category,amount,timestamp
        parts = line.split(',')
        if len(parts) == 4:
            try:
                yield {
                    'category': parts[1].strip(),
                    'amount': float(parts[2].strip())
                }
            except ValueError:
                # Skip malformed rows — in production you would log these
                pass

def run(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument('--input', required=True, help='GCS path to input CSV files')
    parser.add_argument('--output_table', required=True, help='BigQuery table in project:dataset.table format')
    known_args, pipeline_args = parser.parse_known_args(argv)

    options = PipelineOptions(pipeline_args)

    with beam.Pipeline(options=options) as p:
        (
            p
            | 'Read CSV' >> ReadFromText(known_args.input)
            | 'Parse rows' >> beam.ParDo(ParseTransaction())
            | 'Key by category' >> beam.Map(lambda row: (row['category'], row['amount']))
            | 'Sum by category' >> beam.CombinePerKey(sum)
            | 'Format for BQ' >> beam.Map(
                lambda kv: {'category': kv[0], 'total_revenue': round(kv[1], 2)}
            )
            | 'Write to BigQuery' >> WriteToBigQuery(
                known_args.output_table,
                schema='category:STRING,total_revenue:FLOAT',
                write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
            )
        )

if __name__ == '__main__':
    run()

A few things to note about this code:

• We parse the CSV manually instead of using a built-in CSV source. The built-in one works but I find explicit parsing gives you more control when you inevitably hit a bad row.
• We skip malformed rows silently. In production, you would want to route these to a dead-letter table or at least log them with a counter.
• CombinePerKey is more efficient than GroupByKey + Map for aggregations because it does partial combining on each worker before shuffling.
• WRITE_TRUNCATE replaces the table on each run. Use WRITE_APPEND for daily incremental loads.

Running Locally

Before deploying to GCP, test it locally with the DirectRunner:

python pipeline.py \
    --input gs://my-bucket/sample-transactions.csv \
    --output_table my-project:sales_demo.revenue_by_category

The DirectRunner runs everything in a single process on your machine. It is slow for real data but perfect for catching bugs in your pipeline logic. You can also point --input at a local file if you don’t want to touch GCS yet.

If it runs without errors, you should see the aggregated results in your BigQuery table.

Deploying to Dataflow

Now let us run the same pipeline on Dataflow. You need to pass additional pipeline options to tell Beam to use the Dataflow runner:

python pipeline.py \
    --runner DataflowRunner \
    --project my-gcp-project \
    --region australia-southeast1 \
    --temp_location gs://my-bucket/dataflow-temp/ \
    --staging_location gs://my-bucket/dataflow-staging/ \
    --input gs://my-bucket/sales-transactions/*.csv \
    --output_table my-project:sales_demo.revenue_by_category \
    --job_name sales-revenue-batch-$(date +%s)

Breaking this down:

• --runner DataflowRunner tells Beam to run on GCP’s managed service
• --temp_location and --staging_location point to GCS buckets where Dataflow stores intermediate files and your staged code. These are mandatory.
• --job_name must be unique per run, so appending a timestamp avoids collisions
• The --input glob *.csv tells Dataflow to read all matching files, and it will distribute the reads across workers automatically

Once you run this, head to the Dataflow console in GCP and find your job. You will see a graph showing each step of your pipeline and the worker count. For a small batch job, Dataflow usually spins up 1-2 workers. For large datasets, it autoscales up based on the work.

Things to Watch Out For

Here are a few things I learned the hard way:

Startup time is real. Even a tiny job takes 3-4 minutes just to provision workers. If your pipeline runs in under a minute, Dataflow is not the right tool. Cloud Functions or a Cloud Run job with the data loaded into memory is much faster and cheaper.

Staging files pile up. Dataflow uploads your code and dependencies to the staging bucket before every run. Over weeks, this fills up your bucket. Set a lifecycle rule on your staging bucket to delete objects older than, say, 30 days.

Pipeline options get unwieldy fast. When you have 15 parameters, passing them all on the command line gets messy. For production, put your pipeline options in a JSON config file or use something like a ValueProvider to accept runtime parameters without hardcoding them.

The Dataflow shuffle uses GCS under the hood. When you do a GroupByKey or CombinePerKey, Dataflow writes the intermediate data to the temp location. If that bucket is in a different region from your Dataflow workers, you pay egress costs and add latency. Keep them in the same region.

Error handling by default is not great. By default, if one element fails, the whole pipeline might retry or stall depending on how you handle exceptions. Use dead-letter outputs when processing untrusted data. For critical pipelines, add counters for skipped rows and set up monitoring on the Dataflow job metrics.

VPC and networking can surprise you. If your workers need to talk to a Cloud SQL instance on a private IP or an on-prem database, you need to specify a subnetwork in the pipeline options. Without it, Dataflow uses the default network and your connections fail silently or time out.

Production Considerations

For a demo, the code above is fine. For something running in production, here is what I would change:

• Move config to a separate file or use NestedValueProvider so you can pass table names, thresholds, and other parameters without code changes.
• Add proper dead-letter handling. Instead of silently skipping bad rows, write them to a separate BigQuery table with the raw input and an error message.
• Set max_num_workers. Dataflow autoscaling works well most of the time, but I have seen jobs scale to 50 workers for a 2 GB file because of a skewed key. A reasonable upper bound protects your bill.
• Use Flex Templates. Classic templates require you to stage the pipeline as a JSON file. Flex Templates let you package your pipeline as a Docker image and accept parameters at launch time, which is much nicer for CI/CD.
• Add alerts. Dataflow integrates with Cloud Monitoring. Set up an alert if a job fails or runs longer than expected. Nothing worse than finding out on Monday that Friday’s pipeline has been failing since Saturday.

Wrapping Up

Dataflow is my go-to on GCP when a pipeline is too big for a Cloud Function but I do not want to manage a Spark cluster. The Beam SDK takes a bit of getting used to — the functional style with ParDo, CombinePerKey, and the pipe syntax feels odd if you come from a plain Python background — but once it clicks, writing pipelines becomes fast.

Start with the DirectRunner locally to iterate on your logic, then move to Dataflow when you are confident the pipeline is correct. That split between development and execution is one of the best things about Beam, and it saves you from debugging configuration issues and pipeline logic at the same time.

If you are coming from Spark, give Beam a fair try on a small project before rewriting everything. Some things are more verbose, some are cleaner, and you will figure out where it fits in your stack.