Maintaining and Automating Workloads

BigQuery Cost Control

-- Set maximum bytes billed to prevent expensive queries
-- This query will fail if it would scan more than 10 GB
SELECT user_id, COUNT(*) as events
FROM `project.dataset.events`
WHERE DATE(event_time) = CURRENT_DATE()
GROUP BY 1
OPTIONS (maximum_bytes_billed = 10737418240);  -- 10 GB in bytes

-- Monitor slot usage and query costs via INFORMATION_SCHEMA
SELECT
  user_email,
  COUNT(*) AS query_count,
  ROUND(SUM(total_bytes_billed) / POW(1024, 4), 2) AS total_tb_billed,
  ROUND(SUM(total_bytes_billed) / POW(1024, 4) * 6.25, 2) AS estimated_cost_usd,
  ROUND(AVG(total_slot_ms) / 1000, 1) AS avg_slot_seconds
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
  AND job_type = 'QUERY'
  AND state = 'DONE'
GROUP BY 1
ORDER BY estimated_cost_usd DESC
LIMIT 20;

Dataflow Optimization

Dataproc Cost Savings

• Ephemeral clusters — Create, run job, delete. Use --max-idle for auto-deletion after inactivity.
• Preemptible/Spot VMs — Use for secondary workers (up to 80% savings). Primary workers should be standard VMs.
• Autoscaling policies — Define scale-up/down rules based on YARN metrics. Prevents idle over-provisioning.
• Dataproc Serverless — No cluster management. Pay only for DCU-hours consumed during job execution.
• Enhanced Flexibility Mode (EFM) — Gracefully handles preemptible worker removal without losing in-progress work.

# Dataproc autoscaling policy
gcloud dataproc autoscaling-policies create cost-optimized \
  --region=us-central1 \
  --min-instances=2 \
  --max-instances=20 \
  --scale-up-factor=1.0 \
  --scale-down-factor=1.0 \
  --cooldown-period=120s \
  --scale-up-min-worker-fraction=0.0 \
  --scale-down-min-worker-fraction=0.0

# Apply to cluster
gcloud dataproc clusters create my-cluster \
  --region=us-central1 \
  --autoscaling-policy=cost-optimized \
  --num-secondary-workers=10 \
  --secondary-worker-type=spot \
  --worker-machine-type=n2-standard-8 \
  --max-idle=30m

Storage Cost Optimization

Cloud Composer Operations

Production Composer environments require careful configuration and management:

DAG Design Patterns

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.task_group import TaskGroup
from airflow.utils.dates import days_ago
from datetime import timedelta

# Pattern 1: Task Groups for organizing complex DAGs
with DAG('production_pipeline', start_date=days_ago(1),
         schedule_interval='0 6 * * *', catchup=False) as dag:

    start = EmptyOperator(task_id='start')

    with TaskGroup('ingest') as ingest_group:
        ingest_orders = PythonOperator(task_id='orders', python_callable=ingest_orders_fn)
        ingest_users = PythonOperator(task_id='users', python_callable=ingest_users_fn)
        ingest_products = PythonOperator(task_id='products', python_callable=ingest_products_fn)

    with TaskGroup('transform') as transform_group:
        build_facts = PythonOperator(task_id='facts', python_callable=build_facts_fn)
        build_dims = PythonOperator(task_id='dims', python_callable=build_dims_fn)

    with TaskGroup('validate') as validate_group:
        check_counts = PythonOperator(task_id='row_counts', python_callable=check_counts_fn)
        check_freshness = PythonOperator(task_id='freshness', python_callable=check_freshness_fn)

    notify = PythonOperator(task_id='notify', python_callable=send_notification_fn)

    start >> ingest_group >> transform_group >> validate_group >> notify

# Pattern 2: Branching based on data quality
def decide_branch(**context):
    """Route based on data quality check results."""
    quality_score = context['ti'].xcom_pull(task_ids='quality_check')
    if quality_score >= 0.95:
        return 'proceed_to_production'
    elif quality_score >= 0.80:
        return 'proceed_with_warning'
    else:
        return 'quarantine_data'

branch = BranchPythonOperator(
    task_id='quality_branch',
    python_callable=decide_branch,
)

# Pattern 3: Retry and alerting configuration
default_args = {
    'owner': 'data-eng',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,    # Exponential backoff
    'max_retry_delay': timedelta(minutes=60),
    'email_on_failure': True,
    'email_on_retry': False,
    'email': ['data-eng-alerts@example.com'],
    'sla': timedelta(hours=2),            # SLA alerting
    'execution_timeout': timedelta(hours=4),
}

Scheduled Queries in BigQuery

For simpler automation that does not need DAG orchestration, BigQuery scheduled queries run SQL at specified intervals:

-- Scheduled query: Daily aggregation (runs at 06:00 UTC)
-- Configured via Console, bq CLI, or API

# Create a scheduled query via bq CLI
bq query --use_legacy_sql=false \
  --schedule="every 24 hours" \
  --display_name="Daily Revenue Aggregation" \
  --destination_table=project:dataset.daily_revenue \
  --replace=true \
  '
  SELECT
    DATE(event_time) AS event_date,
    region,
    SUM(revenue) AS total_revenue,
    COUNT(DISTINCT user_id) AS unique_users
  FROM `project.dataset.events`
  WHERE DATE(event_time) = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
  GROUP BY 1, 2
  '

BigQuery Editions

BigQuery Editions provide capacity-based pricing as an alternative to on-demand (per-TB) pricing. You purchase slot capacity with autoscaling:

Reservations and Assignments

Slots are organized using three concepts:

• Capacity commitment — The total slot capacity purchased (baseline). Can be committed (1-yr, 3-yr for discounts) or flex (pay-as-you-go).
• Reservation — A named pool of slots from the capacity commitment. Multiple reservations can be created (e.g., "etl", "analytics", "adhoc").
• Assignment — Maps a project, folder, or organization to a reservation. Queries from assigned projects use the reservation's slots.

-- Create a reservation hierarchy
-- Total capacity: 500 slots (baseline) with burst to 1000

# Create capacity commitment
bq mk --capacity_commitment \
  --project_id=admin-project \
  --location=US \
  --slots=500 \
  --plan=ANNUAL \
  --edition=ENTERPRISE

# Create reservations (split the 500 slots)
bq mk --reservation \
  --project_id=admin-project \
  --location=US \
  --reservation_id=etl_jobs \
  --slots=200 \
  --autoscale_max_slots=400

bq mk --reservation \
  --project_id=admin-project \
  --location=US \
  --reservation_id=analytics \
  --slots=200 \
  --autoscale_max_slots=500

bq mk --reservation \
  --project_id=admin-project \
  --location=US \
  --reservation_id=adhoc \
  --slots=100 \
  --autoscale_max_slots=200

# Assign projects to reservations
bq mk --reservation_assignment \
  --project_id=admin-project \
  --location=US \
  --reservation_id=etl_jobs \
  --assignee_id=etl-project \
  --assignee_type=PROJECT \
  --job_type=QUERY

bq mk --reservation_assignment \
  --project_id=admin-project \
  --location=US \
  --reservation_id=analytics \
  --assignee_id=analytics-project \
  --assignee_type=PROJECT \
  --job_type=QUERY

Cloud Monitoring

# Create a Cloud Monitoring alert for Pub/Sub backlog
gcloud alpha monitoring policies create \
  --display-name="PubSub Backlog Alert" \
  --condition-display-name="High backlog" \
  --condition-filter='resource.type="pubsub_subscription" AND metric.type="pubsub.googleapis.com/subscription/num_undelivered_messages"' \
  --condition-threshold-value=10000 \
  --condition-threshold-duration=300s \
  --condition-threshold-comparison=COMPARISON_GT \
  --notification-channels=projects/my-project/notificationChannels/12345

Cloud Logging

Cloud Logging captures logs from all GCP services. Key log sources for data engineering:

• BigQuery — cloudaudit.googleapis.com/data_access for query logs, bigquery.googleapis.com/data_access for data access
• Dataflow — Worker logs in dataflow.googleapis.com/, job messages, step-level diagnostics
• Cloud Composer — Airflow task logs, scheduler logs, DAG processing logs
• Pub/Sub — Dead-letter topic messages for debugging failed message processing

BigQuery Audit and Usage Analysis

-- Identify most expensive queries in the last 7 days
SELECT
  job_id,
  user_email,
  query,
  total_bytes_billed,
  ROUND(total_bytes_billed / POW(1024, 4), 4) AS tb_billed,
  ROUND(total_bytes_billed / POW(1024, 4) * 6.25, 2) AS cost_usd,
  total_slot_ms,
  TIMESTAMP_DIFF(end_time, start_time, SECOND) AS duration_sec
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
  AND job_type = 'QUERY'
  AND state = 'DONE'
  AND error_result IS NULL
ORDER BY total_bytes_billed DESC
LIMIT 25;

-- Identify tables without partition pruning
SELECT
  referenced_table.project_id,
  referenced_table.dataset_id,
  referenced_table.table_id,
  COUNT(*) AS query_count,
  SUM(total_bytes_billed) AS total_bytes_billed
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT,
  UNNEST(referenced_tables) AS referenced_table
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
  AND total_bytes_billed > 10737418240  -- > 10 GB
GROUP BY 1, 2, 3
ORDER BY total_bytes_billed DESC;

Backup and Recovery

-- BigQuery time travel: Query data as it was 2 hours ago
SELECT *
FROM `project.dataset.events`
FOR SYSTEM_TIME AS OF TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 HOUR)
WHERE DATE(event_time) = '2025-03-01';

-- Restore a deleted table (within 7-day time travel window)
# bq cp project:dataset.events@-3600000 project:dataset.events_restored
# (3600000 ms = 1 hour ago)

-- Create a table snapshot (point-in-time copy without storage cost for shared data)
CREATE SNAPSHOT TABLE `project.dataset.events_snapshot_20250301`
CLONE `project.dataset.events`
FOR SYSTEM_TIME AS OF '2025-03-01 00:00:00 UTC'
OPTIONS (
  expiration_timestamp = TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
);

Failover Strategies

# Dataflow: Gracefully drain a streaming pipeline
gcloud dataflow jobs drain JOB_ID --region=us-central1

# Dataflow: Update a streaming pipeline (preserves state)
python my_pipeline.py \
  --runner=DataflowRunner \
  --update \
  --job_name=my-streaming-job \
  --region=us-central1 \
  --transform_name_mapping='{"OldTransformName": "NewTransformName"}'

# Cloud SQL: Set up HA and cross-region replica
gcloud sql instances create my-db \
  --database-version=POSTGRES_15 \
  --tier=db-custom-4-16384 \
  --region=us-central1 \
  --availability-type=REGIONAL         # HA with standby

gcloud sql instances create my-db-replica \
  --master-instance-name=my-db \
  --region=europe-west1                 # Cross-region DR replica

Automation Decision Matrix