如何从Google Cloud Function调用Google Dataproc作业（含存储桶触发）

Got it, let's walk through how to trigger a Dataproc PySpark job from a Cloud Function when a new file lands in GCS. Here's a step-by-step breakdown that should cover everything you need:

Before diving into code, make sure you have these pieces in place:

• Your target GCS bucket (the one where new files will be uploaded)
• Either a persistent Dataproc cluster (for frequent jobs) or plan to use Serverless Dataproc (for on-demand, cost-efficient one-offs)
• Your PySpark job script uploaded to GCS (e.g., gs://your-bucket/spark-scripts/load-to-bq.py)
• Basic IAM permissions sorted out (we'll cover this in detail next)

Cloud Functions need explicit permissions to interact with Dataproc and related services. Find your Cloud Function's default service account (format: PROJECT_ID@appspot.gserviceaccount.com) in the IAM console and add these roles:

• Dataproc Editor: Lets the function submit and manage Dataproc jobs
• Storage Object Viewer: Allows the function to read metadata about the uploaded GCS file and access your Spark script
• Optional: If your Spark job writes to BigQuery, you can either add BigQuery Data Editor to the Cloud Function service account, or (better practice) assign this role directly to your Dataproc cluster's service account.

We'll use Python here since it's the most common runtime for Cloud Functions. First, create a requirements.txt file to pull in the Dataproc client library:

google-cloud-dataproc>=5.0.0

Then the function code itself—this is triggered by a new file landing in your GCS bucket, extracts the file path, and submits the PySpark job to Dataproc:

import os
from google.cloud import dataproc_v1

def trigger_dataproc_job(event, context):
    """Triggered when a new file is uploaded to GCS"""
    uploaded_file = event
    print(f"Processing new file: {uploaded_file['name']} in bucket {uploaded_file['bucket']}")

    # Pull config from environment variables (avoid hardcoding!)
    PROJECT_ID = os.environ.get('PROJECT_ID', 'your-project-id')
    REGION = os.environ.get('REGION', 'us-central1')
    SPARK_SCRIPT_PATH = os.environ.get('SPARK_SCRIPT_PATH', 'gs://your-bucket/spark-scripts/load-to-bq.py')
    
    # Option 1: Use a persistent Dataproc cluster
    CLUSTER_NAME = os.environ.get('CLUSTER_NAME', 'your-persistent-cluster')
    
    # Initialize Dataproc client
    client = dataproc_v1.JobControllerClient(
        client_options={"api_endpoint": f"{REGION}-dataproc.googleapis.com:443"}
    )

    # Define job details—pass the uploaded file path as an argument to your Spark script
    job_config = {
        "placement": {"cluster_name": CLUSTER_NAME},
        "pyspark_job": {
            "main_python_file_uri": SPARK_SCRIPT_PATH,
            "args": [f"gs://{uploaded_file['bucket']}/{uploaded_file['name']}"]
        }
    }

    # Submit the job
    try:
        operation = client.submit_job_as_operation(
            request={"project_id": PROJECT_ID, "region": REGION, "job": job_config}
        )
        job_response = operation.result()
        print(f"Successfully submitted Dataproc job! Job ID: {job_response.reference.job_id}")
    except Exception as e:
        print(f"Failed to submit job: {str(e)}")
        raise e

Alternative: Use Serverless Dataproc (No Persistent Cluster) #

If you don't want to maintain a running cluster, swap the job_config block above with this to spin up a temporary serverless cluster for the job:

job_config = {
    "placement": {
        "managed_cluster": {
            "cluster_name": "temp-serverless-cluster",
            "config": {
                "master_config": {"machine_type_uri": "n1-standard-2"},
                "worker_config": {"machine_type_uri": "n1-standard-2", "num_instances": 2},
                "temp_bucket": "gs://your-temp-bucket"  # Required for serverless job staging
            }
        }
    },
    "pyspark_job": {
        "main_python_file_uri": SPARK_SCRIPT_PATH,
        "args": [f"gs://{uploaded_file['bucket']}/{uploaded_file['name']}"]
    }
}

Don't forget to set all the environment variables (PROJECT_ID, REGION, etc.) in your Cloud Function's deployment settings—this keeps your code clean and configurable.

1. Deploy the Cloud Function, set the trigger type to Cloud Storage, event type to Finalize/Create, and select your target bucket.
2. Upload a test file to the bucket, then check the Cloud Function logs to confirm the job was submitted.
3. Head to the Dataproc console to monitor the job's status, and verify that your PySpark script successfully loads data into BigQuery.

• Error Handling: Add retries for transient errors (try the tenacity library) to handle temporary Dataproc outages. Set up Cloud Monitoring alerts for failed jobs/functions too.
• Spark Script Input: Make sure your PySpark script accepts the input file path via sys.argv so it knows which file to process.
• Cost Optimization: Serverless Dataproc is ideal for infrequent jobs—you only pay for the cluster's runtime. For frequent jobs, a persistent cluster will save you spin-up time.
• Logging: Enable detailed logging for both Cloud Functions and Dataproc jobs. You can view all logs in Cloud Logging to debug issues quickly.

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