Got it, this is exactly the kind of serverless + ephemeral workload that Google Cloud is built for—you’ll avoid wasting money on idle clusters while still processing logs as soon as they land in GCS. Let’s walk through how to build this end-to-end.

Here’s the core flow we’ll implement:

• A new log file is uploaded to your target GCS bucket, triggering a Cloud Function
• The Cloud Function creates an ephemeral (short-lived) Dataproc cluster
• The cluster runs your log-processing job (Spark, Hive, etc.) using the new log file as input
• Once the job finishes (success or failure), the Cloud Function deletes the cluster to stop incurring costs

1. Set Up the Cloud Function Trigger #

First, create a Cloud Function with a GCS Object Finalize trigger:

• Select your target GCS bucket where logs are uploaded
• Optionally, add a prefix filter (e.g., logs/) to only trigger on files in specific paths, avoiding accidental triggers from unrelated files
• Set the runtime to Python 3.11 (or your preferred language; we’ll use Python for this example)

2. Cloud Function Code to Manage Dataproc #

This code handles cluster creation, job submission, job monitoring, and cluster cleanup. Make sure to add google-cloud-dataproc to your function’s requirements.txt file.

import os
import time
from google.cloud import dataproc_v1
from google.cloud.dataproc_v1.types import JobStatus

def trigger_dataproc_processing(event, context):
    # Extract GCS file details from the trigger event
    gcs_event = event
    bucket = gcs_event['bucket']
    file_path = gcs_event['name']
    full_input_path = f"gs://{bucket}/{file_path}"
    
    # Load environment variables (set these in your Cloud Function config)
    PROJECT_ID = os.environ.get('GCP_PROJECT')
    REGION = os.environ.get('REGION', 'us-central1')
    OUTPUT_BUCKET = os.environ.get('OUTPUT_BUCKET')
    
    # Generate unique names for cluster/job to avoid conflicts
    cluster_suffix = context.event_id[:8]  # Use part of the event ID for uniqueness
    cluster_name = f"ephemeral-log-cluster-{cluster_suffix}"
    job_name = f"log-processing-job-{cluster_suffix}"

    # Initialize Dataproc clients
    cluster_client = dataproc_v1.ClusterControllerClient(
        client_options={"api_endpoint": f"{REGION}-dataproc.googleapis.com:443"}
    )
    job_client = dataproc_v1.JobControllerClient(
        client_options={"api_endpoint": f"{REGION}-dataproc.googleapis.com:443"}
    )

    try:
        # 1. Create the ephemeral Dataproc cluster
        cluster_config = {
            "config": {
                "master_config": {"num_instances": 1, "machine_type_uri": "n1-standard-2"},
                "worker_config": {"num_instances": 2, "machine_type_uri": "n1-standard-2"},
                # Use preemptible workers to cut costs (only if your job is fault-tolerant)
                # "secondary_worker_config": {"num_instances": 2, "machine_type_uri": "n1-standard-2", "is_preemptible": True},
                "software_config": {"image_version": "2.1-debian11"}  # Match your job's runtime
            }
        }

        print(f"Starting cluster creation: {cluster_name}")
        cluster_client.create_cluster(
            request={
                "project_id": PROJECT_ID,
                "region": REGION,
                "cluster_name": cluster_name,
                "cluster": cluster_config
            }
        )
        # Wait for cluster to be fully ready
        cluster_client.wait_for_cluster(
            request={"project_id": PROJECT_ID, "region": REGION, "cluster_name": cluster_name}
        )
        print(f"Cluster {cluster_name} is ready to run jobs.")

        # 2. Submit your log-processing job (adjust this to match your job type)
        # Example: Spark job using a custom JAR; swap with Hive/Pig/Spark SQL as needed
        job_config = {
            "placement": {"cluster_name": cluster_name},
            "spark_job": {
                "main_class": "com.yourteam.LogProcessor",
                "jar_file_uris": [f"gs://{OUTPUT_BUCKET}/jars/log-processor.jar"],
                "args": [full_input_path, f"gs://{OUTPUT_BUCKET}/processed-logs/"]
            }
        }

        print(f"Submitting job: {job_name}")
        job_operation = job_client.submit_job_as_operation(
            request={"project_id": PROJECT_ID, "region": REGION, "job": job_config}
        )
        job_result = job_operation.result()

        # 3. Wait for job completion and check status
        job_status = job_result.status
        while job_status.state not in [JobStatus.State.DONE, JobStatus.State.ERROR, JobStatus.State.CANCELLED]:
            job_result = job_client.get_job(
                request={"project_id": PROJECT_ID, "region": REGION, "job_id": job_result.job_id}
            )
            job_status = job_result.status
            time.sleep(5)

        if job_status.state == JobStatus.State.ERROR:
            error_msg = f"Job failed with details: {job_status.details}"
            print(error_msg)
            raise Exception(error_msg)
        
        print(f"Job {job_name} completed successfully! Output saved to {OUTPUT_BUCKET}/processed-logs/")

    finally:
        # 4. Delete the cluster NO MATTER what (avoids orphaned clusters costing money)
        print(f"Cleaning up cluster: {cluster_name}")
        cluster_client.delete_cluster(
            request={"project_id": PROJECT_ID, "region": REGION, "cluster_name": cluster_name}
        )
        print(f"Cluster {cluster_name} deleted.")

3. Required IAM Permissions #

Your Cloud Function’s default service account needs these roles to operate:

• Dataproc Cluster Admin (roles/dataproc.clusterAdmin): Create and delete clusters
• Dataproc Job Admin (roles/dataproc.jobAdmin): Submit and monitor jobs
• Storage Object Viewer (roles/storage.objectViewer): Read the input log files from GCS
• Storage Object Creator (roles/storage.objectCreator): Write processed output to your target GCS bucket

4. Key Optimizations for Your Use Case #

Since you only get a few logs per day, these tweaks will make the workflow more efficient and cost-effective:

• Avoid duplicate triggers: If multiple logs upload at once, use Cloud Pub/Sub as a middle layer—forward GCS events to Pub/Sub, then have your Cloud Function subscribe with batching enabled. This lets you process multiple logs in a single cluster run.
• Use preemptible VMs: For worker nodes, enable preemptible instances (commented out in the code above) to cut costs by ~70%. Only do this if your log processing is idempotent (can be re-run without issues).
• Custom Dataproc images: Build a custom image with your job dependencies pre-installed to reduce cluster startup time.
• Monitor costs: Set up Cloud Billing alerts to notify you if Dataproc costs exceed your threshold, just in case clusters aren’t deleted properly.

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