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.

$1

``mermaid

graph TB

subgraph "Job Configuration"

A["Container Image"]

B["Resource Limits"]

C["Environment"]

end

subgraph "Execution"

D["Task Instances"]

E["Parallel Tasks"]

F["Retries"]

end

subgraph "Monitoring"

G["Logs"]

H["Metrics"]

I["Status"]

end

A --> D

B --> D

C --> D

D --> G

E --> H

F --> I

classDef gcp fill:#1a73e8,stroke:#fff,color:#fff

class A,B,C,D,E,F,G,H,I gcp

`

$1

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

$1

$1

`yaml

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"

`

$1

`python

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()

`

$1

$1

`python

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)

`

$1

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

$1

$1

`python

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

`

$1

`python

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}")

`

$1

$1

`python

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],

)

`

$1

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

$1

$1

1. Idempotency

`python

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

`python

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

``

$1

$1

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

$1

1. [Cloud Run Jobs Documentation](https://cloud.google.com/run/docs/create-jobs)

2. [Job Configuration Guide](https://cloud.google.com/run/docs/configuring/jobs)

3. [Monitoring and Logging](https://cloud.google.com/run/docs/monitoring)

4. [Best Practices](https://cloud.google.com/run/docs/best-practices)

5. [Error Handling](https://cloud.google.com/run/docs/troubleshooting)

6. [Parallel Processing](https://cloud.google.com/run/docs/parallel)

$1

[GCP Vertex AI](/posts/gcp/vertex-ai) - ML workloads

[GCP Cost Optimization](/posts/gcp/cost-optimization) - Optimize costs

[GCP vs AWS](/posts/gcp/vs-aws-2024) - Cloud comparison

[GCP Serverless](/posts/gcp/serverless) - Serverless architecture