使用Apache Beam Python API从GCP MySQL导出至GCS Bucket遇问题求助

Hey there, let's break down and fix the issues you're facing with your Apache Beam pipeline that moves data from GCP Cloud SQL (MySQL) to GCS. I've worked through similar problems before, so let's tackle each error one by one and get your pipeline running properly.

First, Let's Recap Your Core Issues #

You hit four main roadblocks:

• No Dataflow job was created on your first run, just partial logs
• "Credential file not found" error after setting GOOGLE_APPLICATION_CREDENTIALS via shell script
• Parameter parsing getting dropped unexpectedly
• 403 Permission Denied errors when making requests to GCP services

Fix 1: Credential File Not Found & Parameter Parsing Issues #

Credential Path Mistakes #

The most common reason for this error is using a relative path instead of an absolute path for your service account key. Let's fix that:

# ❌ Bad: Relative paths can break if your working directory changes
export GOOGLE_APPLICATION_CREDENTIALS="my-key.json"

# ✅ Good: Use an absolute path to ensure the file is always found
export GOOGLE_APPLICATION_CREDENTIALS="/home/your-user/gcp-projects/my-service-key.json"

Also, double-check the file exists and has proper read permissions:

chmod 400 /absolute/path/to/your-key.json

Fixing Dropped Parameters #

Beam expects command-line arguments to be passed directly after your script, without wrapping them in unnecessary quotes. Here's the correct way to structure your run command (or shell script):

python mysql_to_gcs.py \
  --runner DataflowRunner \
  --project your-gcp-project-id \
  --region us-central1 \
  --temp_location gs://your-bucket/temp \
  --staging_location gs://your-bucket/staging \
  --mysql_host your-cloudsql-ip \
  --mysql_user your-db-username \
  --mysql_password your-db-password \
  --mysql_db your-db-name \
  --output_path gs://your-bucket/output/mysql-data-*

If you're using a shell script, make sure each argument is on its own line (the backslashes are important for line continuation) and avoid wrapping groups of arguments in quotes.

Fix 2: 403 Permission Denied Errors #

Your service account needs specific permissions to interact with Dataflow, Cloud SQL, and GCS. Let's assign the necessary roles:

1. Dataflow Worker: Lets the service account run Dataflow jobs
2. Cloud SQL Client: Allows connecting to your Cloud SQL MySQL instance
3. Storage Object Creator: Permits writing files to your GCS bucket
4. Storage Object Viewer: Lets Dataflow read from temp/staging buckets

You can add these roles via the GCP IAM console, or use the gcloud CLI:

# Add Dataflow Worker role
gcloud projects add-iam-policy-binding your-project-id \
  --member="serviceAccount:your-service-account@your-project-id.iam.gserviceaccount.com" \
  --role="roles/dataflow.worker"

# Add Cloud SQL Client role
gcloud projects add-iam-policy-binding your-project-id \
  --member="serviceAccount:your-service-account@your-project-id.iam.gserviceaccount.com" \
  --role="roles/cloudsql.client"

# Add Storage Object Creator role
gcloud projects add-iam-policy-binding your-project-id \
  --member="serviceAccount:your-service-account@your-project-id.iam.gserviceaccount.com" \
  --role="roles/storage.objectCreator"

# Add Storage Object Viewer role (for temp/staging buckets)
gcloud projects add-iam-policy-binding your-project-id \
  --member="serviceAccount:your-service-account@your-project-id.iam.gserviceaccount.com" \
  --role="roles/storage.objectViewer"

Fix 3: Corrected Pipeline Code #

Here's a polished version of your pipeline that addresses common pitfalls, includes proper parameter parsing, and works with GCP Dataflow:

import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.io.jdbc import ReadFromJdbc

def run():
    # Parse command-line arguments
    parser = argparse.ArgumentParser(description='Apache Beam Pipeline: MySQL to GCS')
    parser.add_argument('--mysql_host', required=True, help='Cloud SQL MySQL instance public IP/hostname')
    parser.add_argument('--mysql_user', required=True, help='MySQL database username')
    parser.add_argument('--mysql_password', required=True, help='MySQL database password')
    parser.add_argument('--mysql_db', required=True, help='MySQL database name')
    parser.add_argument('--output_path', required=True, help='GCS output path (e.g., gs://bucket/prefix)')
    known_args, pipeline_args = parser.parse_known_args()

    # Configure Pipeline options for Dataflow
    pipeline_options = PipelineOptions(pipeline_args)
    standard_options = pipeline_options.view_as(StandardOptions)
    standard_options.runner = 'DataflowRunner'

    # JDBC configuration for MySQL (update the query to match your table!)
    jdbc_config = {
        'driver_class_name': 'com.mysql.cj.jdbc.Driver',
        'jdbc_url': f'jdbc:mysql://{known_args.mysql_host}:3306/{known_args.mysql_db}?useSSL=false',
        'username': known_args.mysql_user,
        'password': known_args.mysql_password,
        'query': 'SELECT * FROM your_target_table;',  # Replace with your actual query
    }

    # Build and run the pipeline
    with beam.Pipeline(options=pipeline_options) as p:
        # Read data from MySQL
        mysql_records = p | 'Read from MySQL' >> ReadFromJdbc(**jdbc_config)

        # Convert records to CSV format (adjust this if you need a different output)
        formatted_records = mysql_records | 'Format to CSV' >> beam.Map(
            lambda row: ','.join(str(value) for value in row.values())
        )

        # Write formatted data to GCS
        formatted_records | 'Write to GCS' >> beam.io.WriteToText(
            file_path_prefix=known_args.output_path,
            file_name_suffix='.csv',
            shard_name_template='-SSSSS'  # Adds shard number to output files
        )

if __name__ == '__main__':
    run()

Important Notes for This Code: #

• Replace your_target_table with the actual table you want to export
• Ensure your Cloud SQL instance has a public IP, and you've whitelisted Dataflow's IP range (or use VPC peering/Serverless VPC Access for better security)
• Install required dependencies first:

  pip install apache-beam[gcp] mysql-connector-python
  

Fix 4: No Dataflow Job Created (Only Partial Logs) #

This usually happens for one of these reasons:

1. You didn't specify --runner DataflowRunner: By default, Beam uses DirectRunner which runs locally (no Dataflow job created)
2. Missing required Dataflow parameters: You need to pass --project, --region, --temp_location, and --staging_location to submit a job to Dataflow
3. Code syntax errors: The pipeline crashes before it can submit the job to GCP

Quick Checks: #

• Add --verbose to your run command to get more detailed logs
• Validate your code syntax before running:

  python -m py_compile mysql_to_gcs.py
  

• Make sure all required Dataflow parameters are included in your run command (like the example in Fix 1)

Final Step-by-Step Run Process #

1. Set your service account credential:

   export GOOGLE_APPLICATION_CREDENTIALS="/absolute/path/to/your-service-key.json"
   

2. Install dependencies:

   pip install apache-beam[gcp] mysql-connector-python
   

3. Run the pipeline with all required parameters:

   python mysql_to_gcs.py \
     --runner DataflowRunner \
     --project your-gcp-project-id \
     --region us-central1 \
     --temp_location gs://your-bucket/temp \
     --staging_location gs://your-bucket/staging \
     --mysql_host 12.34.56.78 \  # Replace with your Cloud SQL public IP
     --mysql_user root \
     --mysql_password your-db-password \
     --mysql_db your-db-name \
     --output_path gs://your-bucket/output/mysql-data
   

内容的提问来源于stack exchange，提问作者Arpan Roy