Cloud Functions Architecture

Cloud Functions is a serverless execution environment for building and connecting cloud services. For data engineers, it's ideal for lightweight, event-driven data processing tasks.

Gen 1 vs Gen 2

Deployment Example

# Deploy Gen 2 Cloud Function for data processing
gcloud functions deploy process-gcs-upload \
  --gen2 \
  --runtime=python311 \
  --region=us-central1 \
  --source=./functions \
  --entry-point=process_upload \
  --trigger-event-filters="type=google.cloud.storage.object.v1.finalized" \
  --trigger-event-filters="bucket=my-data-lake" \
  --memory=1024MB \
  --timeout=300s \
  --min-instances=0 \
  --max-instances=100 \
  --service-account=data-pipeline@project.iam.gserviceaccount.com \
  --set-env-vars=BIGQUERY_DATASET=raw_data,PROJECT_ID=my-project

Event-Driven Data Processing

GCS Upload Trigger

import functions_framework
from google.cloud import bigquery
from google.cloud import storage
import json
import base64

@functions_framework.cloud_event
def process_gcs_upload(cloud_event):
    """Process files uploaded to GCS and load into BigQuery."""
    data = cloud_event.data

    bucket_name = data["bucket"]
    file_name = data["name"]
    file_size = data["size"]

    print(f"Processing file: {file_name} ({file_size} bytes) from {bucket_name}")

    # Determine file type and process accordingly
    if file_name.endswith(".json"):
        process_json_file(bucket_name, file_name)
    elif file_name.endswith(".parquet"):
        process_parquet_file(bucket_name, file_name)
    elif file_name.endswith(".csv"):
        process_csv_file(bucket_name, file_name)

def process_json_file(bucket_name, file_name):
    """Load JSON file into BigQuery."""
    client = bigquery.Client()

    # Configure load job
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
        time_partitioning=bigquery.TimePartitioning(
            type_=bigquery.TimePartitioningType.DAY,
            field="event_date"
        ),
        clustering_fields=["event_type", "user_id"]
    )

    uri = f"gs://{bucket_name}/{file_name}"

    load_job = client.load_table_from_uri(
        uri,
        "project.dataset.events",
        job_config=job_config
    )

    load_job.result()  # Wait for job to complete

    table = client.get_table("project.dataset.events")
    print(f"Loaded {table.num_rows} rows into events table")

Pub/Sub Trigger for Data Processing

import functions_framework
from google.cloud import bigquery
import json

@functions_framework.cloud_event
def process_pubsub_message(cloud_event):
    """Process Pub/Sub message and load into BigQuery."""
    # Decode Pub/Sub message
    message_data = base64.b64decode(
        cloud_event.data["message"]["data"]
    ).decode("utf-8")

    # Parse JSON payload
    payload = json.loads(message_data)

    # Validate and transform data
    transformed_data = transform_event(payload)

    # Load into BigQuery
    client = bigquery.Client()

    errors = client.insert_rows_json(
        "project.dataset.real_time_events",
        [transformed_data]
    )

    if errors:
        print(f"Insert errors: {errors}")
        raise Exception(f"BigQuery insert failed: {errors}")

    print(f"Successfully processed event: {transformed_data['event_id']}")

def transform_event(event):
    """Transform raw event for BigQuery."""
    return {
        "event_id": event.get("id"),
        "event_type": event.get("type"),
        "user_id": event.get("user_id"),
        "timestamp": event.get("timestamp"),
        "payload": json.dumps(event.get("data", {})),
        "processed_at": datetime.utcnow().isoformat()
    }

Cloud Functions for Data Pipeline Orchestration

Triggering Dataflow Jobs

import functions_framework
from google.cloud import dataflow_v1beta3

@functions_framework.http
def trigger_dataflow_job(request):
    """Trigger a Dataflow batch job via HTTP request."""
    dataflow_client = dataflow_v1beta3.DataflowV1Beta3Client()

    request_body = dataflow_v1beta3.LaunchTemplateParameters(
        job_name=f"batch-job-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}",
        parameters={
            "input": "gs://my-data-lake/raw/daily/",
            "output": "project.dataset.processed",
            "date": datetime.utcnow().strftime('%Y-%m-%d')
        },
        environment=dataflow_v1beta3.RuntimeEnvironment(
            num_workers=4,
            machine_type="n1-standard-4",
            temp_location="gs://my-data-lake/temp/",
            zone="us-central1-a"
        )
    )

    response = dataflow_client.launch_template(
        project_id="my-project",
        location="us-central1",
        gcs_path="gs://my-templates/dataflow-batch-pipeline",
        launch_parameters=request_body
    )

    return {
        "status": "success",
        "job_id": response.job_id,
        "job_name": request_body.job_name
    }

Scheduling Batch Processing

import functions_framework
from google.cloud import scheduler_v1
from google.cloud import bigquery

@functions_framework.cloud_event
def daily_aggregation(cloud_event):
    """Run daily aggregation queries in BigQuery."""
    client = bigquery.Client()

    # Run aggregation query
    query = """
    CREATE OR REPLACE TABLE `project.dataset.daily_summary`
    PARTITION BY DATE(event_date)
    CLUSTER BY event_type
    AS
    SELECT
        DATE(event_timestamp) as event_date,
        event_type,
        COUNT(*) as event_count,
        COUNT(DISTINCT user_id) as unique_users,
        AVG(payload_value) as avg_value,
        SUM(payload_value) as total_value
    FROM `project.dataset.events`
    WHERE DATE(event_timestamp) = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
    GROUP BY 1, 2
    """

    query_job = client.query(query)
    query_job.result()

    print(f"Aggregation complete: {query_job.num_dml_affected_rows} rows processed")

Memory and Timeout Configuration

Cost Optimization

# Cloud Functions pricing (us-central1)
pricing = {
    "invocations": {
        "free_tier": 2_000_000,  # 2M invocations/month
        "cost_per_100k": 0.40    # $0.40 per 100K invocations
    },
    "compute": {
        "free_tier_gb_seconds": 400_000,  # 400K GB-seconds/month
        "cost_per_gb_second": 0.0000025   # $0.0000025 per GB-second
    },
    "networking": {
        "egress_free": "First 5GB/month free",
        "egress_cost": "$0.12/GB after free tier"
    }
}

# Cost optimization strategies
optimization = {
    "batch_processing": "Process multiple records per invocation",
    "connection_pooling": "Reuse BigQuery clients across invocations",
    "memory_tuning": "Right-size memory to avoid over-provisioning",
    "min_instances": "Use 0 for cost savings, >0 for latency",
    "timeout_tuning": "Set appropriate timeouts to avoid unnecessary charges"
}

Common Interview Questions

Q1: When would you use Cloud Functions vs. Cloud Run for data processing?

Answer: Cloud Functions is ideal for lightweight, event-driven tasks (<9 min execution, <2GB memory). Use it for GCS upload processing, Pub/Sub message handling, and simple transformations. Cloud Run is better for longer-running processes, containerized workloads, and when you need more control over the runtime environment.

Q2: How do you handle failures in Cloud Functions data processing?

Answer: Implement retry logic with exponential backoff for transient failures. Use Pub/Sub dead-letter queues for messages that fail repeatedly. Log errors to Cloud Logging for monitoring. Set up Cloud Monitoring alerts for failure rates. For critical pipelines, implement circuit breaker patterns.

Q3: What are the limitations of Cloud Functions for data engineering?

Answer: Key limitations: 9-minute timeout (Gen 1) / 60 minutes (Gen 2), 2GB memory (Gen 1) / 32GB (Gen 2), 1000 concurrent invocations per region, cold start latency (200ms-2s). For large-scale data processing, consider Dataflow or Dataproc.

Q4: How do you optimize Cloud Functions for cost?

Answer: 1) Batch records to reduce invocations, 2) Right-size memory (don't over-provision), 3) Use min_instances=0 for batch workloads, 4) Implement connection pooling for database clients, 5) Use faster runtimes (Go, Node.js) for simpler functions, 6) Cache frequently accessed data.

Q5: How do you secure Cloud Functions in a data pipeline?

Answer: Use service accounts with minimal required permissions. Enable VPC Connector for private network access. Use Secret Manager for credentials. Implement IAM authentication for HTTP-triggered functions. Enable audit logging. Set up Cloud Armor for DDoS protection on public endpoints.