Parallel & Serverless CSV Ingestion to Cloud SQL Using Cloud Dataflow

Anyone who has tried bulk-loading hundreds of gigabytes of multi-file CSV exports from Google Cloud Storage (GCS) into Cloud SQL for PostgreSQL hits the same frustrating wall: gcloud sql import csv invokes the Cloud SQL Admin API, which serializes imports one operation at a time per instance.

If your upstream pipeline lands 200 partitioned CSV files in GCS, importing them sequentially via Admin API calls leaves your database’s vCPUs and disk IOPS largely underutilized while blowing past batch SLA windows. Conversely, writing standard row-by-row JDBC INSERT batches from a distributed runner saturates WAL logging and round-trip latency.

To get the best of both worlds—horizontal serverless parallelism plus native PostgreSQL bulk load speed—I built an open-source Apache Beam / Google Cloud Dataflow Flex Template that streams partitioned CSV bundles concurrently into Cloud SQL PostgreSQL via COPY ... FROM STDIN (github.com/datasherlock/csql-dataflow-pgcopy-connector).

Why PostgreSQL COPY FROM STDIN + Dataflow Wins

• Bypasses Admin API Serialization: Dataflow workers open standard database sessions in parallel up to your controlled worker concurrency, whereas Cloud SQL Admin imports lock at concurrency = 1.
• Orders of Magnitude Faster Than JDBC INSERT: PostgreSQL’s COPY ... FROM STDIN parses raw delimited streams directly into heap pages with minimal per-row executor and network overhead.
• Partition-Aligned Writes Eliminate Lock Contention: By extracting the target PostgreSQL table’s partition key inside Apache Beam and grouping bundles per partition before calling COPY, concurrent workers load disjoint child partitions without contending on the same B-tree pages.

Inside the Apache Beam Pipeline (csql_write_pipeline_pgcopy.py)

The pipeline matches wildcard GCS URIs (gs://bucket/path/*.csv), chunks reads into 1 GB bundles via ReadAllFromText, extracts the target partition key per record, groups records by partition, and streams each partition group through an in-memory buffer into PostgreSQL’s cursor.copy_expert("COPY ... FROM STDIN", buffer):

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from common.utils import parse_args
from sinks.write_to_cloudsql_pgcopy import CopyCsvToPostgres
from sources.csv import ExtractPartitionKey

def run():
    parser = parse_args()
    args, pipeline_args = parser.parse_known_args()
    pipeline_args.extend([f"--runner={args.runner}"])

    options = PipelineOptions(pipeline_args)
    with beam.Pipeline(options=options) as p:
        (
            p
            | "MatchFiles" >> beam.io.fileio.MatchFiles(args.source_path.strip('"'))
            | "ReadMatches" >> beam.io.fileio.ReadMatches()
            | "ExtractFilePaths" >> beam.Map(lambda element: element.metadata.path)
            | "ChunkData" >> beam.io.textio.ReadAllFromText(
                desired_bundle_size=1073741824,  # 1 GB chunks per worker bundle
                skip_header_lines=0,
            )
            | "ExtractPartitionKey" >> beam.ParDo(ExtractPartitionKey(args))
            | "GroupByPartitionKey" >> beam.GroupByKey()
            | "CopyToPostgres" >> beam.ParDo(CopyCsvToPostgres(args))
        )

Production Tuning Guardrails for Cloud SQL

1. Cap --max_num_workers to Cloud SQL vCPUs: Never unleash 100 Dataflow workers against an 8-vCPU Cloud SQL instance! Start with worker concurrency at 0.5x–1.5x your Cloud SQL vCPU count and monitor CPU, active connections, and disk write throughput.
2. Drop Non-Essential Secondary Indexes Before Initial Backfills: Just like traditional relational bulk loads, building B-tree indexes once via parallel CREATE INDEX after COPY completes is dramatically faster than maintaining secondary indexes per row.
3. Package as a Dataflow Flex Template: Using the included Dockerfile in the repo, publish the pipeline once to Artifact Registry + GCS so Cloud Composer (Airflow) or Cloud Scheduler can trigger parameterized bulk loads on demand.

Originally published in the Google Cloud Community on Medium: Parallel & Serverless CSV Ingestion to Cloud SQL Using Cloud Dataflow. Full Flex Template code: github.com/datasherlock/csql-dataflow-pgcopy-connector.