Google Cloud Run Jobs: Complete Guide to Serverless Batch Processing

Cloud Run Jobs provides a serverless platform for executing batch workloads and background processing tasks. This guide explores how to leverage Cloud Run Jobs for efficient and scalable batch processing.

Architecture Overview

Flowchart Diagram

100%

Copy

Rendering diagram...

Key Features

Feature	Description	Benefits
Serverless	No infrastructure management	Simplified operations
Parallel Execution	Multiple task instances	Faster processing
Retry Logic	Automatic retries	Reliability
Monitoring	Built-in observability	Visibility

Implementation Examples

Basic Job Configuration

# job.yaml
apiVersion: run.googleapis.com/v1
kind: Job
metadata:
  name: batch-processor
spec:
  template:
    spec:
      containers:
      - image: gcr.io/project-id/processor:latest
        resources:
          limits:
            cpu: "1"
            memory: "2Gi"
        env:
        - name: BATCH_SIZE
          value: "100"
        - name: INPUT_BUCKET
          value: "gs://input-data"
        - name: OUTPUT_BUCKET
          value: "gs://processed-data"

Parallel Job Processing

from google.cloud import run_v2
from google.cloud.run_v2 import Job, TaskTemplate

def create_parallel_job(
    project: str,
    location: str,
    job_name: str,
    image_uri: str,
    task_count: int = 10,
):
    client = run_v2.JobsClient()
    parent = f"projects/{project}/locations/{location}"
    
    job = Job(
        template=TaskTemplate(
            containers=[{
                "image": image_uri,
                "resources": {
                    "limits": {
                        "cpu": "1",
                        "memory": "2Gi"
                    }
                }
            }],
            max_retries=3,
            task_count=task_count,
        ),
        labels={"environment": "production"}
    )
    
    operation = client.create_job(
        parent=parent,
        job_id=job_name,
        job=job
    )
    
    return operation.result()

Job Development

Container Implementation

import os
import time
from google.cloud import storage

def process_batch():
    # Get environment variables
    batch_size = int(os.getenv("BATCH_SIZE", "100"))
    input_bucket = os.getenv("INPUT_BUCKET")
    output_bucket = os.getenv("OUTPUT_BUCKET")
    task_index = int(os.getenv("CLOUD_RUN_TASK_INDEX", "0"))
    task_count = int(os.getenv("CLOUD_RUN_TASK_COUNT", "1"))
    
    # Initialize storage client
    storage_client = storage.Client()
    
    # Calculate batch range for this task
    start_index = task_index * batch_size
    end_index = start_index + batch_size
    
    try:
        # Process batch
        process_items(start_index, end_index)
        return True
    except Exception as e:
        print(f"Error processing batch: {e}")
        return False

if __name__ == "__main__":
    success = process_batch()
    exit(0 if success else 1)

Job Configuration Options

Option	Value	Description
CPU	1-8 cores	Processing power
Memory	512MB-32GB	Available RAM
Timeout	Up to 24h	Maximum duration

Job Management

Job Execution

from google.cloud import run_v2

def execute_job(
    project: str,
    location: str,
    job_name: str,
):
    client = run_v2.JobsClient()
    
    request = run_v2.RunJobRequest(
        name=f"projects/{project}/locations/{location}/jobs/{job_name}"
    )
    
    operation = client.run_job(request=request)
    execution = operation.result()
    
    print(f"Job execution completed with status: {execution.status}")
    return execution

def monitor_execution(
    project: str,
    location: str,
    job_name: str,
    execution_name: str,
):
    client = run_v2.JobsClient()
    
    request = run_v2.GetExecutionRequest(
        name=f"projects/{project}/locations/{location}/jobs/{job_name}/executions/{execution_name}"
    )
    
    while True:
        execution = client.get_execution(request=request)
        print(f"Execution status: {execution.status}")
        
        if execution.status in ["SUCCEEDED", "FAILED"]:
            break
            
        time.sleep(10)
    
    return execution

Error Handling

def handle_job_failure(execution):
    if execution.status == "FAILED":
        # Get task details
        for task in execution.tasks:
            if task.status == "FAILED":
                print(f"Task {task.index} failed:")
                print(f"Exit code: {task.exit_code}")
                print(f"Error: {task.status_message}")
                
                # Get logs for failed task
                log_name = f"projects/{project_id}/logs/run.googleapis.com%2Fjob-execution"
                filter_str = f'resource.labels.job_name="{job_name}" AND resource.labels.task_index="{task.index}"'
                
                logging_client = logging.Client()
                for entry in logging_client.list_entries(filter_=filter_str):
                    print(f"Log: {entry.payload}")

Monitoring and Logging

Metrics Configuration

from google.cloud import monitoring_v3

def setup_monitoring(
    project: str,
    job_name: str,
):
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{project}"
    
    # Create custom metric descriptor
    descriptor = monitoring_v3.MetricDescriptor(
        type_="custom.googleapis.com/cloud_run_jobs/batch_processing",
        metric_kind=monitoring_v3.MetricDescriptor.MetricKind.GAUGE,
        value_type=monitoring_v3.MetricDescriptor.ValueType.INT64,
        description="Batch processing metrics for Cloud Run Jobs",
    )
    
    client.create_metric_descriptor(
        name=project_name,
        metric_descriptor=descriptor,
    )
    
    return descriptor

def record_metric(
    project: str,
    job_name: str,
    value: int,
):
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{project}"
    
    series = monitoring_v3.TimeSeries()
    series.metric.type = "custom.googleapis.com/cloud_run_jobs/batch_processing"
    series.metric.labels["job_name"] = job_name
    
    point = series.points.add()
    point.value.int64_value = value
    point.interval.end_time.seconds = int(time.time())
    
    client.create_time_series(
        name=project_name,
        time_series=[series],
    )

Performance Metrics

Metric	Target	Action if Exceeded
Task Duration	< 30 minutes	Optimize code
Memory Usage	< 80%	Increase limit
Error Rate	< 1%	Debug issues

Best Practices

Job Design Patterns

1. Idempotency

   def process_item(item_id: str):
       # Generate deterministic execution ID
       execution_id = hashlib.sha256(f"{item_id}".encode()).hexdigest()
       
       # Check if already processed
       if is_processed(execution_id):
           return
       
       try:
           # Process item
           result = perform_processing(item_id)
           
           # Mark as processed
           mark_processed(execution_id, result)
       except Exception as e:
           log_error(execution_id, e)
           raise
   

2. Checkpointing

   def process_batch_with_checkpoint():
       checkpoint_bucket = storage.Client().bucket("checkpoints")
       checkpoint_blob = checkpoint_bucket.blob(f"job-{JOB_ID}/checkpoint")
       
       # Load last checkpoint
       last_processed = 0
       if checkpoint_blob.exists():
           last_processed = int(checkpoint_blob.download_as_text())
       
       try:
           # Process items from last checkpoint
           for i in range(last_processed, TOTAL_ITEMS):
               process_item(i)
               
               # Save checkpoint every 100 items
               if i % 100 == 0:
                   checkpoint_blob.upload_from_string(str(i))
       except Exception as e:
           # Save checkpoint on failure
           checkpoint_blob.upload_from_string(str(last_processed))
           raise
   

Troubleshooting Guide

Common Issues

Issue	Cause	Solution
Memory Errors	Large batches	Reduce batch size
Timeouts	Long processing	Split tasks
Concurrency Issues	Race conditions	Use locks

References

1. Cloud Run Jobs Documentation
2. Job Configuration Guide
3. Monitoring and Logging
4. Best Practices
5. Error Handling
6. Parallel Processing

Related Posts

• GCP Vertex AI - ML workloads
• GCP Cost Optimization - Optimize costs
• GCP vs AWS - Cloud comparison
• GCP Serverless - Serverless architecture