如何用Cloud Dataflow Python SDK每日将GCS中CSV批量追加至BigQuery

Got it, let's break down exactly how to build a daily Cloud Dataflow pipeline (using the Python SDK) to load hundreds of CSV files from GCS and append them to a single BigQuery table. I'll walk you through the setup, code, and best practices below:

Before diving into code, make sure you have these covered:

• A Google Cloud project with the Dataflow, BigQuery, and GCS APIs enabled
• The Apache Beam GCP SDK installed locally: pip install apache-beam[gcp]
• A target BigQuery table (either pre-created with the correct schema, or we can let Dataflow create it for you—pre-creating is safer to avoid schema mismatches)

The pipeline will follow this simple flow:

1. Read all target CSV files from GCS (using wildcards to match hundreds of files)
2. Parse CSV rows into structured data that matches your BigQuery table schema
3. Append the parsed data to your BigQuery table
4. Set up a daily trigger to run this pipeline automatically

Here's a complete, commented code sample you can adapt to your use case:

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions
from apache_beam.io.gcp.bigquery import WriteToBigQuery, BigQueryDisposition
from apache_beam.io.csv import ReadFromCsv  # Use this for easier header/schema handling (Beam 2.40+)

def run_daily_csv_to_bq():
    # Configure pipeline base options
    pipeline_options = PipelineOptions()
    
    # Set GCP-specific settings
    gcp_options = pipeline_options.view_as(GoogleCloudOptions)
    gcp_options.project = "your-gcp-project-id"
    gcp_options.job_name = "daily-csv-to-bq-load"
    gcp_options.staging_location = "gs://your-bucket-name/staging"
    gcp_options.temp_location = "gs://your-bucket-name/temp"
    
    # Choose runner: use DirectRunner for local testing, DataflowRunner for production
    pipeline_options.view_as(StandardOptions).runner = "DataflowRunner"
    
    # BigQuery target details
    bq_table_id = "your-project-id:your-dataset-name.your-target-table"
    # Define schema if your table isn't pre-created (match CSV columns exactly)
    bq_table_schema = "user_id:STRING, transaction_amount:FLOAT, transaction_date:TIMESTAMP, region:STRING"

    with beam.Pipeline(options=pipeline_options) as p:
        # Step 1: Read CSV files from GCS (wildcard matches all CSVs in the directory)
        # Use ReadFromCsv if your CSVs have headers for automatic schema matching
        csv_rows = p | "Read CSV Files" >> ReadFromCsv(
            file_pattern="gs://your-bucket-name/csv-uploads/*.csv",
            skip_header_lines=1,  # Skip the first header row in each CSV
            schema=bq_table_schema
        )

        # Alternative: If you need custom parsing (e.g., handling quoted commas), use ReadFromText + custom map
        # def parse_csv_line(line):
        #     columns = line.split(",")
        #     # Add type conversions and cleanup here
        #     return {
        #         "user_id": columns[0].strip(),
        #         "transaction_amount": float(columns[1].strip()),
        #         "transaction_date": columns[2].strip(),
        #         "region": columns[3].strip()
        #     }
        # csv_rows = p | "Read Raw CSV" >> beam.io.ReadFromText("gs://your-bucket/csv-uploads/*.csv", skip_header_lines=1) | "Parse CSV" >> beam.Map(parse_csv_line)

        # Step 2: Append parsed data to BigQuery
        csv_rows | "Write to BigQuery" >> WriteToBigQuery(
            table=bq_table_id,
            schema=bq_table_schema,
            write_disposition=BigQueryDisposition.WRITE_APPEND,  # Critical: ensures we append, not overwrite
            create_disposition=BigQueryDisposition.CREATE_IF_NEEDED  # Only create table if it doesn't exist
        )

if __name__ == "__main__":
    run_daily_csv_to_bq()

• Wildcard Flexibility: Use *.csv to match all files in a directory, or **/*.csv to recurse through subdirectories. Dataflow automatically parallelizes reading across hundreds of files.
• Schema Consistency: Double-check that your CSV column order and data types match the BigQuery table. Mismatches (e.g., a string in an integer column) will cause failed writes.
• Daily Trigger Setup: Use Cloud Scheduler to run this pipeline daily:
  1. Package your code into a Dataflow template (run gcloud dataflow templates build to create it)
  2. Create a Cloud Scheduler job that runs gcloud dataflow jobs run with your template and parameters
• Error Handling: Add a dead-letter queue for failed rows using beam.Partition to separate bad data and write it to a GCS bucket for later debugging.
• Performance: For large datasets, adjust max_num_workers in your pipeline options to enable auto-scaling, and set min_bundle_size on ReadFromText to optimize parallel processing.

Before deploying to production, test with the DirectRunner (change the runner in the code) using a small set of CSV files. This lets you validate parsing and BigQuery writes without incurring Dataflow costs.

内容的提问来源于stack exchange，提问作者Parth Desai