Your Spark cluster is running. Your Airflow DAGs are green. Your stakeholders are happy with the dashboards. And then the cloud bill arrives.
It's $47,000 for a single month — up 34% from last quarter — and nobody can quite explain why. The infrastructure team points at the data team. The data team points at the analytics team. Everyone agrees the number is too high. Nobody agrees on which pipeline is responsible, which query is wasteful, or whether it's even worth optimizing versus just raising the budget ceiling. This is not a hypothetical. This exact situation plays out at hundreds of organizations every year, and the root cause is almost never a single runaway query. It's the accumulated weight of a hundred reasonable-at-the-time decisions made without visibility into their downstream cost consequences.
By the end of this lesson, you'll have the mental models and practical tooling to actually fix this. We're going to build a systematic approach to cost attribution at the pipeline level — not just "which GCP project spent the most," but which specific DAG, which task, which transformation, which partition strategy is responsible for your spend. Then we'll work through concrete optimization techniques backed by profiling data, covering compute, storage, and query costs across AWS, GCP, and Azure ecosystems.
What you'll learn:
You should be comfortable with:
You don't need deep DevOps or FinOps expertise — we'll build that understanding here.
Most cloud cost dashboards give you service-level attribution: BigQuery spent $12,400 this month, EC2 spent $18,200, S3 spent $3,100. This is nearly useless for a data team trying to optimize spend. You need to know that the customer_lifetime_value dbt model is scanning 2.8 TB every time it runs because nobody added a partition filter, and it runs six times a day.
The gap between "service-level spend" and "pipeline-level spend" is the attribution problem. Solving it requires three things:
1. A tagging strategy that survives execution. Every compute resource, every query, every storage operation needs to carry metadata that links it back to the pipeline context that created it. This sounds obvious, but most teams tag their infrastructure (the cluster) and not their workloads (the jobs running on the cluster).
2. A cost allocation model that handles shared resources. A Spark cluster might run ten different jobs during its lifetime. A data warehouse might serve both production pipelines and ad-hoc analyst queries. You need a model for distributing costs across these consumers proportionally.
3. Telemetry that captures resource consumption at the right granularity. Cloud billing data arrives in daily or hourly granularity, aggregated by service. You need per-run, per-task metrics that let you correlate cost with pipeline behavior.
Let's build each of these pieces.
The most important design decision is: tag at the workload level, not just the infrastructure level. Here's what this means concretely across different environments.
In Airflow, you want every task to propagate context labels to whatever cloud resource it touches:
# dags/customer_pipeline.py
from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from datetime import datetime
PIPELINE_LABELS = {
"pipeline": "customer-pipeline",
"team": "data-engineering",
"cost-center": "analytics-platform",
"environment": "production",
"data-domain": "customer",
}
with DAG(
dag_id="customer_lifetime_value",
start_date=datetime(2024, 1, 1),
schedule_interval="0 */4 * * *", # every 4 hours
tags=["customer", "production", "revenue"],
) as dag:
compute_clv = BigQueryInsertJobOperator(
task_id="compute_clv_model",
configuration={
"query": {
"query": "{% include 'sql/clv_model.sql' %}",
"useLegacySql": False,
"jobTimeoutMs": 300000,
# Labels propagate to BigQuery job metadata and appear in billing export
"labels": {
**PIPELINE_LABELS,
"task": "compute-clv-model",
"dag-run": "{{ run_id }}",
},
}
},
project_id="my-data-project",
location="US",
)
The labels dictionary here is the critical piece. BigQuery exposes these labels in the INFORMATION_SCHEMA.JOBS view and in the billing export, which means you can later run:
-- BigQuery: cost breakdown by pipeline task over the last 30 days
SELECT
labels.value AS task_name,
COUNT(*) AS job_count,
ROUND(SUM(total_bytes_billed) / POW(10, 12) * 6.25, 2) AS estimated_cost_usd,
ROUND(AVG(total_bytes_billed) / POW(10, 9), 1) AS avg_gb_billed,
MAX(TIMESTAMP_DIFF(end_time, creation_time, SECOND)) AS max_duration_seconds
FROM
`region-us`.INFORMATION_SCHEMA.JOBS,
UNNEST(labels) AS labels
WHERE
labels.key = 'task'
AND DATE(creation_time) >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
AND job_type = 'QUERY'
GROUP BY
labels.value
ORDER BY
estimated_cost_usd DESC;
This query alone, run for the first time on a real production environment, tends to produce some memorable moments.
For Spark on EMR or Dataproc, the equivalent mechanism is Spark application tags and cluster labels:
# spark_job_config.py
import boto3
from datetime import datetime
def submit_spark_job(pipeline_name: str, task_name: str, script_s3_path: str):
"""Submit a tagged Spark job to EMR with full cost attribution metadata."""
emr_client = boto3.client("emr", region_name="us-east-1")
# These tags appear in AWS Cost Explorer with a 24-hour lag
job_tags = [
{"Key": "Pipeline", "Value": pipeline_name},
{"Key": "Task", "Value": task_name},
{"Key": "Team", "Value": "data-engineering"},
{"Key": "CostCenter", "Value": "analytics-platform"},
{"Key": "RunDate", "Value": datetime.utcnow().strftime("%Y-%m-%d")},
{"Key": "Environment", "Value": "production"},
]
response = emr_client.run_job_flow(
Name=f"{pipeline_name}-{task_name}-{datetime.utcnow().strftime('%Y%m%d%H%M')}",
ReleaseLabel="emr-7.0.0",
Applications=[{"Name": "Spark"}],
Instances={
"InstanceGroups": [
{
"Name": "Master",
"Market": "ON_DEMAND",
"InstanceRole": "MASTER",
"InstanceType": "m5.xlarge",
"InstanceCount": 1,
},
{
"Name": "Core",
"Market": "SPOT",
"InstanceRole": "CORE",
"InstanceType": "m5.4xlarge",
"InstanceCount": 4,
"BidPrice": "0.30",
},
],
"KeepJobFlowAliveWhenNoSteps": False,
"TerminationProtected": False,
},
Steps=[
{
"Name": task_name,
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": [
"spark-submit",
"--deploy-mode", "cluster",
"--conf", f"spark.yarn.tags={pipeline_name},{task_name}",
script_s3_path,
],
},
}
],
JobFlowRole="EMR_EC2_DefaultRole",
ServiceRole="EMR_DefaultRole",
Tags=job_tags,
AutoTerminationPolicy={"IdleTimeout": 3600},
)
return response["JobFlowId"]
Warning: AWS tag propagation to Cost Explorer is not instantaneous. Tags applied at cluster creation appear in Cost Explorer within 24 hours, but tags added after cluster creation may take up to 72 hours to show up in billing data. Always tag at creation time.
When pipelines share infrastructure, you need a proportional attribution model. The right metric depends on what you're allocating:
Here's a practical allocation model for a shared Spark cluster, using the Spark History Server REST API to pull per-application metrics:
# cost_attribution/spark_allocator.py
import requests
from dataclasses import dataclass
from typing import Dict, List
import json
@dataclass
class SparkApplicationMetrics:
app_id: str
app_name: str
duration_ms: int
executor_cpu_time_ns: int
peak_executor_memory_bytes: int
total_input_bytes: int
total_output_bytes: int
total_shuffle_read_bytes: int
total_shuffle_write_bytes: int
def fetch_spark_app_metrics(history_server_url: str) -> List[SparkApplicationMetrics]:
"""Pull application-level metrics from Spark History Server."""
apps_response = requests.get(
f"{history_server_url}/api/v1/applications",
params={"status": "completed", "limit": 200}
)
apps = apps_response.json()
metrics = []
for app in apps:
app_id = app["id"]
# Get executor-level stats for this application
executors_response = requests.get(
f"{history_server_url}/api/v1/applications/{app_id}/executors"
)
executors = executors_response.json()
# Aggregate across all executors (excluding driver)
worker_executors = [e for e in executors if e["id"] != "driver"]
total_cpu_ns = sum(e.get("totalCores", 0) * app.get("duration", 0) * 1_000_000
for e in worker_executors)
peak_mem = max((e.get("peakMemoryMetrics", {}).get("JVMHeapMemory", 0)
for e in worker_executors), default=0)
total_input = sum(e.get("totalInputBytes", 0) for e in worker_executors)
total_output = sum(e.get("totalShuffleWrite", 0) for e in worker_executors)
metrics.append(SparkApplicationMetrics(
app_id=app_id,
app_name=app.get("name", "unknown"),
duration_ms=app.get("duration", 0),
executor_cpu_time_ns=total_cpu_ns,
peak_executor_memory_bytes=peak_mem,
total_input_bytes=total_input,
total_output_bytes=total_output,
total_shuffle_read_bytes=sum(e.get("totalShuffleRead", 0) for e in worker_executors),
total_shuffle_write_bytes=total_output,
))
return metrics
def allocate_cluster_cost(
metrics: List[SparkApplicationMetrics],
total_cluster_cost_usd: float,
weight_cpu: float = 0.5,
weight_memory: float = 0.3,
weight_io: float = 0.2,
) -> Dict[str, float]:
"""
Distribute cluster cost across applications using a weighted resource model.
The weighting philosophy: CPU time is the primary driver of compute cost,
memory is secondary (affects instance sizing), and I/O determines
storage and network costs but is already partially captured by duration.
"""
total_cpu = sum(m.executor_cpu_time_ns for m in metrics)
total_mem = sum(m.peak_executor_memory_bytes for m in metrics)
total_io = sum(m.total_input_bytes + m.total_output_bytes for m in metrics)
# Guard against division by zero
total_cpu = total_cpu or 1
total_mem = total_mem or 1
total_io = total_io or 1
attribution = {}
for m in metrics:
cpu_share = (m.executor_cpu_time_ns / total_cpu) * weight_cpu
mem_share = (m.peak_executor_memory_bytes / total_mem) * weight_memory
io_share = ((m.total_input_bytes + m.total_output_bytes) / total_io) * weight_io
combined_share = cpu_share + mem_share + io_share
attribution[m.app_name] = round(combined_share * total_cluster_cost_usd, 4)
return attribution
This gives you a defensible, tunable cost attribution that you can feed into your internal chargeback or showback model.
Cost attribution tells you who is spending. Profiling tells you why. The distinction matters enormously because the fix for "this pipeline costs $800/run" depends entirely on whether the cost comes from data volume, execution time, inefficient queries, or unnecessary re-computation.
BigQuery's INFORMATION_SCHEMA is criminally underused. Most teams check it to see if a job failed. You should be using it to build a continuous cost profile of your entire query workload.
-- Identify the top 20 most expensive recurring queries over the past 7 days
-- This query itself is ~0 cost because INFORMATION_SCHEMA queries don't bill for data scanned
WITH query_fingerprints AS (
SELECT
-- Normalize the query to group parameterized variants together
REGEXP_REPLACE(
REGEXP_REPLACE(query, r'\d{4}-\d{2}-\d{2}', '<DATE>'),
r"'[^']*'", "'<STRING>'"
) AS query_template,
total_bytes_billed,
total_bytes_processed,
TIMESTAMP_DIFF(end_time, start_time, MILLISECOND) AS duration_ms,
user_email,
labels,
job_id,
creation_time,
-- Cache hit means $0 cost, always worth tracking
cache_hit,
FROM
`region-us`.INFORMATION_SCHEMA.JOBS
WHERE
DATE(creation_time) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
AND job_type = 'QUERY'
AND state = 'DONE'
AND error_result IS NULL
AND cache_hit = FALSE -- only count billable queries
),
aggregated AS (
SELECT
query_template,
COUNT(*) AS execution_count,
ROUND(SUM(total_bytes_billed) / POW(10, 12) * 6.25, 2) AS total_cost_usd,
ROUND(AVG(total_bytes_billed) / POW(10, 9), 1) AS avg_gb_billed,
MAX(total_bytes_billed / POW(10, 9)) AS max_gb_single_run,
ROUND(AVG(duration_ms) / 1000, 1) AS avg_duration_seconds,
-- How much would we save if queries ran at their minimum observed cost?
ROUND(
(SUM(total_bytes_billed) - MIN(total_bytes_billed) * COUNT(*)) / POW(10, 12) * 6.25,
2
) AS potential_savings_from_optimization_usd,
FROM
query_fingerprints
GROUP BY
query_template
HAVING
execution_count >= 3 -- only patterns, not one-offs
)
SELECT *
FROM aggregated
ORDER BY total_cost_usd DESC
LIMIT 20;
The potential_savings_from_optimization_usd column is particularly powerful: it tells you how much you'd save if every run of a query behaved like its cheapest observed run. If this number is high relative to the total cost, it means the query cost is highly variable — which usually indicates missing partition filters that only get applied sometimes, or a data skew problem.
The Spark UI is the richest profiling tool available to you, but most engineers only look at whether stages completed. Here's what you should actually be examining.
The Stages tab: Sort by "Duration" descending. Any stage that took more than 3x the median stage duration is a red flag for data skew.
The SQL tab: Look for the "Exchange" nodes in the query plan — these are shuffle operations, and they're expensive both in time and I/O. If you see a SortMergeJoin where a BroadcastHashJoin should be, that's costing you significantly.
The Executors tab: Compare the task time distribution across executors. Ideal: narrow distribution, high utilization. Bad: a few executors with 10x the task time of others (skew), or many executors sitting idle (over-provisioning).
Here's a script to pull Spark stage metrics programmatically and flag anomalies:
# cost_attribution/spark_profiler.py
import requests
from typing import List, Dict, Optional
import statistics
def profile_spark_stages(
history_server_url: str,
app_id: str,
skew_threshold: float = 3.0,
spill_threshold_mb: float = 500.0,
) -> Dict:
"""
Profile a Spark application's stages and surface cost-relevant anomalies.
Returns a structured report highlighting skew, spill, and shuffle costs.
"""
stages_url = f"{history_server_url}/api/v1/applications/{app_id}/stages"
stages = requests.get(stages_url).json()
issues = []
stage_summaries = []
for stage in stages:
if stage.get("status") != "COMPLETE":
continue
stage_id = stage["stageId"]
# Get per-task metrics for this stage
tasks_url = f"{history_server_url}/api/v1/applications/{app_id}/stages/{stage_id}/0/taskList"
tasks_response = requests.get(tasks_url, params={"length": 2000})
tasks = tasks_response.json()
if not tasks:
continue
task_durations = [t.get("taskMetrics", {}).get("executorRunTime", 0) for t in tasks]
task_input_bytes = [t.get("taskMetrics", {}).get("inputMetrics", {}).get("bytesRead", 0)
for t in tasks]
if not task_durations or max(task_durations) == 0:
continue
median_duration = statistics.median(task_durations)
max_duration = max(task_durations)
# Detect skew: any task taking >3x the median
skew_ratio = max_duration / median_duration if median_duration > 0 else 1
# Detect spill: tasks writing to disk because they ran out of memory
spill_bytes = sum(
t.get("taskMetrics", {}).get("memoryBytesSpilled", 0) for t in tasks
)
spill_mb = spill_bytes / (1024 ** 2)
# Shuffle read/write
shuffle_read = sum(
t.get("taskMetrics", {}).get("shuffleReadMetrics", {}).get("totalBytesRead", 0)
for t in tasks
)
shuffle_write = sum(
t.get("taskMetrics", {}).get("shuffleWriteMetrics", {}).get("bytesWritten", 0)
for t in tasks
)
summary = {
"stage_id": stage_id,
"stage_name": stage.get("name", ""),
"num_tasks": len(tasks),
"duration_seconds": stage.get("executorRunTime", 0) / 1000,
"skew_ratio": round(skew_ratio, 2),
"spill_mb": round(spill_mb, 1),
"shuffle_read_mb": round(shuffle_read / (1024**2), 1),
"shuffle_write_mb": round(shuffle_write / (1024**2), 1),
"input_gb": round(sum(task_input_bytes) / (1024**3), 2),
}
stage_summaries.append(summary)
if skew_ratio > skew_threshold:
issues.append({
"type": "DATA_SKEW",
"stage_id": stage_id,
"stage_name": stage.get("name", ""),
"severity": "HIGH" if skew_ratio > 10 else "MEDIUM",
"detail": f"Max task duration {max_duration/1000:.1f}s is {skew_ratio:.1f}x the median {median_duration/1000:.1f}s",
"recommendation": "Check for skewed join keys or skewed partitions. Consider salting or AQE skew handling.",
})
if spill_mb > spill_threshold_mb:
issues.append({
"type": "MEMORY_SPILL",
"stage_id": stage_id,
"severity": "HIGH" if spill_mb > 5000 else "MEDIUM",
"detail": f"Stage spilled {spill_mb:.0f} MB to disk",
"recommendation": (
"Increase executor memory, reduce partition size via repartition(), "
"or use broadcast joins to avoid large shuffles."
),
})
return {
"app_id": app_id,
"stage_summaries": sorted(stage_summaries, key=lambda x: x["duration_seconds"], reverse=True),
"issues": issues,
"total_shuffle_gb": round(
sum(s["shuffle_read_mb"] + s["shuffle_write_mb"] for s in stage_summaries) / 1024, 2
),
}
Tip: Shuffle volume is one of the most reliable proxies for Spark job cost in cloud environments. Every shuffle byte has to be written to disk and read back, which translates directly to IOPS and I/O costs. A job that shuffles 500 GB is likely 5-10x more expensive than an equivalent job that shuffles 50 GB.
Once you have profiling data, the same problems appear repeatedly. Here are the five that generate the most cloud spend, with concrete remediation strategies.
This is the single most common source of unexpected BigQuery and Athena costs. A table is partitioned by event_date, but a query accidentally applies a non-pushdown filter that forces a full scan.
-- EXPENSIVE: This scans every partition because the CAST happens before the filter
-- BigQuery will scan the entire table
SELECT *
FROM `analytics.events`
WHERE CAST(event_timestamp AS DATE) = '2024-11-15'
-- CHEAP: This pushes the filter down to the partition pruner
-- BigQuery scans only the matching partition
SELECT *
FROM `analytics.events`
WHERE DATE(event_timestamp) = '2024-11-15'
-- OR, even more explicitly:
WHERE event_date = '2024-11-15' -- if event_date is the partition column
The issue here is subtle: BigQuery's partition pruner recognizes DATE() as a safe transformation for timestamp partition columns but doesn't always recognize CAST(). The practical difference can be 100x in cost.
In dbt, always validate that your partition filters propagate to the underlying SQL:
# models/marts/customer_events.yml
models:
- name: customer_events
config:
materialized: incremental
partition_by:
field: event_date
data_type: date
granularity: day
# This filter must reference the partition column exactly
incremental_strategy: insert_overwrite
partitions_to_replace: >
[
date_sub(current_date, interval 3 day),
date_sub(current_date, interval 2 day),
date_sub(current_date, interval 1 day),
current_date
]
-- models/marts/customer_events.sql
SELECT
event_date,
user_id,
event_type,
properties,
created_at
FROM {{ ref('stg_raw_events') }}
{% if is_incremental() %}
-- The partition filter must use the exact partition column name
WHERE event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 DAY)
{% endif %}
This one is terrifying because it can turn a $5 query into a $500 query silently. A seemingly innocent join against a reference table that was accidentally duplicated produces a cartesian product.
-- This looks innocent. It's not.
-- If dim_products somehow has duplicate product_ids (from a CDC issue, etc.),
-- this join fans out every row in orders
SELECT
o.order_id,
o.customer_id,
p.product_name,
p.category
FROM orders o
JOIN dim_products p ON o.product_id = p.product_id
WHERE o.order_date >= '2024-01-01'
-- SAFER: Always validate join cardinality on reference tables
-- Run this before joining:
SELECT product_id, COUNT(*) as cnt
FROM dim_products
HAVING cnt > 1;
-- If this returns rows, you have a dimension integrity problem
-- that will silently corrupt and inflate your queries
In dbt, you can enforce this with relationship tests, but the more important habit is pre-join validation in your transformation code:
# pyspark: defensive join with fan-out detection
from pyspark.sql import DataFrame
import pyspark.sql.functions as F
def safe_join(
left: DataFrame,
right: DataFrame,
join_key: str,
join_type: str = "left",
expected_max_cardinality: int = 1,
) -> DataFrame:
"""
Join two DataFrames with fan-out detection.
Raises a ValueError if the right-side DataFrame has duplicates on the join key,
preventing accidental cartesian product scenarios.
"""
right_cardinality = (
right
.groupBy(join_key)
.count()
.agg(F.max("count").alias("max_count"))
.collect()[0]["max_count"]
)
if right_cardinality > expected_max_cardinality:
raise ValueError(
f"Join key '{join_key}' has max cardinality {right_cardinality} in right DataFrame "
f"(expected <= {expected_max_cardinality}). "
f"This join would produce a fan-out. Deduplicate before joining."
)
return left.join(right, on=join_key, how=join_type)
In BigQuery, sharded tables (the _ suffix pattern like events_20241101) are a legacy pattern, but they're still common in older data stacks and in organizations that migrated from Google Analytics. The wildcard query pattern is seductive and dangerous:
-- This scans every shard. If you have 3 years of data, this is expensive.
SELECT * FROM `analytics.events_*`
WHERE event_type = 'purchase'
-- This scans only the shards in range using the _TABLE_SUFFIX pseudo-column
SELECT * FROM `analytics.events_*`
WHERE _TABLE_SUFFIX BETWEEN '20241101' AND '20241130'
AND event_type = 'purchase'
The broader lesson: understand what your query engine's partition pruning actually guarantees. Every engine has edge cases where filters don't push down as expected.
This is a compute problem rather than a query problem. A Spark cluster configured for peak load that runs 22 hours of the day at 5% utilization is one of the most wasteful patterns in data engineering.
The fix requires understanding your workload's actual resource consumption profile. Pull utilization data from CloudWatch, Stackdriver, or your Kubernetes metrics server:
# cost_attribution/cluster_profiler.py
import boto3
from datetime import datetime, timedelta
from typing import Dict, List
import statistics
def profile_emr_cluster_utilization(
cluster_id: str,
lookback_hours: int = 168, # 1 week
) -> Dict:
"""
Profile an EMR cluster's actual vs. provisioned resource utilization.
Returns utilization statistics and a right-sizing recommendation.
"""
cw = boto3.client("cloudwatch", region_name="us-east-1")
emr = boto3.client("emr", region_name="us-east-1")
cluster = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=lookback_hours)
def get_metric(metric_name: str, stat: str = "Average") -> List[float]:
response = cw.get_metric_statistics(
Namespace="AWS/ElasticMapReduce",
MetricName=metric_name,
Dimensions=[{"Name": "JobFlowId", "Value": cluster_id}],
StartTime=start_time,
EndTime=end_time,
Period=3600, # 1-hour granularity
Statistics=[stat],
)
return [dp[stat] for dp in sorted(response["Datapoints"], key=lambda x: x["Timestamp"])]
yarn_memory_available = get_metric("YARNMemoryAvailablePercentage")
container_pending = get_metric("ContainerPendingRatio")
hdfs_utilization = get_metric("HDFSUtilization")
if not yarn_memory_available:
return {"error": "No metrics available for this cluster"}
avg_memory_available = statistics.mean(yarn_memory_available)
p95_container_pending = sorted(container_pending)[int(len(container_pending) * 0.95)] if container_pending else 0
# Right-sizing logic:
# If average available memory > 40%, cluster is over-provisioned
# If p95 pending ratio > 0.5, cluster is under-provisioned at peak
recommendation = None
if avg_memory_available > 40 and p95_container_pending < 0.1:
# Safe to downsize
suggested_reduction = min(0.4, (avg_memory_available - 20) / 100)
recommendation = {
"action": "DOWNSIZE",
"current_avg_memory_available_pct": round(avg_memory_available, 1),
"suggested_core_reduction_pct": round(suggested_reduction * 100, 0),
"rationale": (
f"Average available memory {avg_memory_available:.1f}% with negligible pending ratio. "
f"Safe to reduce core nodes by ~{suggested_reduction*100:.0f}%."
),
}
elif p95_container_pending > 0.5:
recommendation = {
"action": "UPSIZE_OR_OPTIMIZE",
"p95_pending_ratio": round(p95_container_pending, 2),
"rationale": (
"High p95 pending ratio indicates resource contention. Consider adding core nodes "
"OR reducing job concurrency before scaling up hardware."
),
}
else:
recommendation = {"action": "MAINTAIN", "rationale": "Cluster utilization is within acceptable range."}
return {
"cluster_id": cluster_id,
"cluster_name": cluster.get("Name"),
"avg_memory_available_pct": round(avg_memory_available, 1),
"p95_container_pending_ratio": round(p95_container_pending, 2),
"recommendation": recommendation,
}
This is the incremental processing problem. A pipeline that recomputes all-time customer spend from raw events every day is performing O(N) work when it should perform O(delta) work.
The fix is incremental materialization, but it has to be implemented correctly. The most common mistake is incremental logic that still scans the full source table to find new records:
-- BAD incremental pattern: scans entire source table every run
-- This is O(N) regardless of how much new data arrived
{{ config(materialized='incremental') }}
SELECT
customer_id,
SUM(order_amount) as lifetime_value,
COUNT(*) as order_count
FROM {{ ref('orders') }} -- FULL TABLE SCAN every run
{% if is_incremental() %}
WHERE order_date > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}
GROUP BY customer_id
-- GOOD incremental pattern: source scan is bounded by new data
-- This requires your source table to be partitioned on order_date
{{ config(
materialized='incremental',
partition_by={'field': 'order_date', 'data_type': 'date'},
incremental_strategy='merge',
unique_key='customer_id'
) }}
WITH new_orders AS (
SELECT *
FROM {{ ref('orders') }}
WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 DAY) -- partition-pruned scan
),
incremental_aggregation AS (
SELECT
customer_id,
SUM(order_amount) as period_value,
COUNT(*) as period_order_count,
MAX(order_date) as last_order_date
FROM new_orders
GROUP BY customer_id
)
SELECT * FROM incremental_aggregation
The MERGE strategy in dbt will then upsert these results into the existing aggregation table, adding new customers and updating existing ones — without rescanning historical data.
Compute gets all the attention, but storage costs are often larger and stealthier. A few patterns that add up:
Raw data that's more than 30 days old almost never needs to be in hot storage. Implement lifecycle policies that automatically move data to cheaper tiers:
# storage_lifecycle/s3_lifecycle_manager.py
import boto3
import json
def apply_data_lake_lifecycle_policy(bucket_name: str, prefix: str = "data/"):
"""
Apply a tiered lifecycle policy to a data lake S3 bucket.
Tier logic:
- 0-30 days: S3 Standard (hot, immediate access)
- 30-90 days: S3 Standard-IA (infrequent access, 40% cheaper)
- 90-365 days: S3 Glacier Instant Retrieval (rare access, 68% cheaper)
- 365+ days: S3 Glacier Deep Archive (archival, 95% cheaper)
"""
s3 = boto3.client("s3")
lifecycle_config = {
"Rules": [
{
"ID": f"data-lake-tiering-{prefix.replace('/', '-')}",
"Status": "Enabled",
"Filter": {"Prefix": prefix},
"Transitions": [
{"Days": 30, "StorageClass": "STANDARD_IA"},
{"Days": 90, "StorageClass": "GLACIER_IR"},
{"Days": 365, "StorageClass": "DEEP_ARCHIVE"},
],
"NoncurrentVersionTransitions": [
{"NoncurrentDays": 7, "StorageClass": "STANDARD_IA"},
{"NoncurrentDays": 30, "StorageClass": "GLACIER_IR"},
],
"NoncurrentVersionExpiration": {
"NoncurrentDays": 90 # Delete old versions after 90 days
},
}
]
}
s3.put_bucket_lifecycle_configuration(
Bucket=bucket_name,
LifecycleConfiguration=lifecycle_config,
)
print(f"Applied lifecycle policy to s3://{bucket_name}/{prefix}")
The choice of file format has enormous cost implications in query engines that bill by data scanned. Parquet with Snappy compression is the standard recommendation, but the details matter:
# pyspark: writing optimally partitioned, compressed Parquet
from pyspark.sql import SparkSession, DataFrame
import pyspark.sql.functions as F
def write_optimized_parquet(
df: DataFrame,
output_path: str,
partition_cols: list,
target_file_size_mb: int = 256,
):
"""
Write a DataFrame to Parquet with optimal file size and compression.
The target_file_size_mb of 256MB is the sweet spot for BigQuery and Athena:
- Too small (< 64MB): excessive file overhead, slow metadata operations
- Too large (> 512MB): poor parallelism on read, slow single-file scans
"""
# Estimate row count and data size to calculate optimal partition count
row_count = df.count()
# Sample to estimate average row size in bytes (Parquet, compressed)
sample_size = min(10000, row_count)
# Repartition to hit our target file size
# Heuristic: average Parquet compression ratio for mixed data is ~5:1
# Adjust target_bytes_per_partition accordingly
target_bytes_per_file = target_file_size_mb * 1024 * 1024
# Sort by partition columns before writing to maximize run-length encoding
# This can reduce file size by 20-60% on high-cardinality string columns
sorted_df = df.sortWithinPartitions(*partition_cols)
(
sorted_df
.write
.mode("overwrite")
.option("compression", "snappy")
# Limit files per partition to avoid small file problem
.option("maxRecordsPerFile", 500_000)
.partitionBy(*partition_cols)
.parquet(output_path)
)
Warning:
sortWithinPartitionssorts within each Spark partition, not globally. If you need global sort order for optimal Parquet statistics (which dramatically speeds up predicate pushdown), useorderBy()instead — but be aware this triggers a global sort, which is a full shuffle and significantly more expensive. UsesortWithinPartitionsfor compression optimization,orderByonly when you need the Parquet min/max statistics to be globally accurate.
Fixing costs once is easy. Preventing them from returning without adding bureaucratic overhead is the hard problem.
Build a system that flags new queries or pipelines that exhibit expensive patterns before they hit production:
# cost_governance/query_guard.py
import re
from dataclasses import dataclass
from typing import List
@dataclass
class CostWarning:
severity: str # "BLOCK" | "WARN" | "INFO"
rule: str
message: str
def analyze_query_for_cost_risks(query: str, dialect: str = "bigquery") -> List[CostWarning]:
"""
Static analysis of SQL queries for common cost anti-patterns.
This is intentionally not a complete SQL parser — it's a fast heuristic
check suitable for CI/CD pipelines. False positives are acceptable.
False negatives are not.
"""
warnings = []
query_upper = query.upper().strip()
# Rule 1: SELECT * on large-known tables
if re.search(r'SELECT\s+\*\s+FROM', query_upper):
warnings.append(CostWarning(
severity="WARN",
rule="WILDCARD_SELECT",
message="SELECT * scans all columns. Specify only required columns to reduce bytes billed.",
))
# Rule 2: Missing WHERE clause entirely (on non-aggregation queries)
is_aggregation = any(kw in query_upper for kw in ["GROUP BY", "COUNT(", "SUM(", "AVG("])
if "WHERE" not in query_upper and not is_aggregation:
warnings.append(CostWarning(
severity="WARN",
rule="MISSING_FILTER",
message="Query has no WHERE clause. Verify this full scan is intentional.",
))
# Rule 3: CROSS JOIN
if re.search(r'CROSS\s+JOIN', query_upper):
warnings.append(CostWarning(
severity="BLOCK",
rule="EXPLICIT_CROSS_JOIN",
message="CROSS JOIN detected. This produces a cartesian product. Use INNER/LEFT JOIN with explicit conditions.",
))
# Rule 4: Wildcard table without _TABLE_SUFFIX filter (BigQuery specific)
if dialect == "bigquery":
has_wildcard_table = re.search(r'FROM\s+`[^`]+\*`', query_upper)
has_suffix_filter = "_TABLE_SUFFIX" in query_upper
if has_wildcard_table and not has_suffix_filter:
warnings.append(CostWarning(
severity="BLOCK",
rule="WILDCARD_TABLE_NO_SUFFIX_FILTER",
message=(
"Wildcard table query without _TABLE_SUFFIX filter will scan all shards. "
"Add: WHERE _TABLE_SUFFIX BETWEEN 'YYYYMMDD' AND 'YYYYMMDD'"
),
))
# Rule 5: CAST on partition column (anti-pattern for partition pruning)
if re.search(r'WHERE.*CAST\([^)]+AS\s+DATE\)', query_upper):
warnings.append(CostWarning(
severity="WARN",
rule="CAST_ON_PARTITION_COLUMN",
message=(
"CAST() on a timestamp in WHERE clause may prevent partition pruning. "
"Use DATE() or compare against a typed literal instead."
),
))
return warnings
def enforce_cost_policy(query: str, dialect: str = "bigquery", fail_on_block: bool = True) -> bool:
"""
Run cost policy enforcement. Returns True if query passes, False if blocked.
Suitable for integration into dbt pre-hooks or CI pipeline checks.
"""
warnings = analyze_query_for_cost_risks(query, dialect)
blocked = False
for w in warnings:
prefix = "🚫 BLOCKED" if w.severity == "BLOCK" else "⚠️ WARNING" if w.severity == "WARN" else "ℹ️ INFO"
print(f"{prefix} [{w.rule}]: {w.message}")
if w.severity == "BLOCK":
blocked = True
if blocked and fail_on_block:
raise ValueError("Query blocked by cost policy. Fix the issues above before proceeding.")
return not blocked
Integrate this into your dbt project as a pre-hook or into your CI pipeline as a PR check. The BLOCK severity should genuinely block deployment; WARN should require an override comment in the PR.
Set up programmatic budget alerts that fire before you hit your ceiling, not after:
# cost_governance/budget_alerts.py
import boto3
import json
def create_pipeline_budget_alert(
pipeline_name: str,
monthly_budget_usd: float,
alert_thresholds: List[float] = [0.5, 0.8, 1.0, 1.2],
notification_emails: List[str] = None,
):
"""
Create an AWS Budgets alert for a specific pipeline using Cost Allocation Tags.
Requires that your pipeline resources are tagged with 'Pipeline' = pipeline_name.
Tag-based budgets take 24-48h to activate after tagging.
"""
budgets = boto3.client("budgets", region_name="us-east-1")
account_id = boto3.client("sts").get_caller_identity()["Account"]
notifications = []
for threshold in alert_thresholds:
is_over = threshold >= 1.0
notifications.append({
"Notification": {
"NotificationType": "ACTUAL" if is_over else "FORECASTED",
"ComparisonOperator": "GREATER_THAN",
"Threshold": threshold * 100, # AWS expects percentage
"ThresholdType": "PERCENTAGE",
"NotificationState": "ALARM",
},
"Subscribers": [
{"SubscriptionType": "EMAIL", "Address": email}
for email in (notification_emails or [])
],
})
budgets.create_budget(
AccountId=account_id,
Budget={
"BudgetName": f"pipeline-{pipeline_name}-monthly",
"BudgetLimit": {"Amount": str(monthly_budget_usd), "Unit": "USD"},
"TimeUnit": "MONTHLY",
"BudgetType": "COST",
"CostFilters": {
"TagKeyValue": [f"user:Pipeline${pipeline_name}"]
},
},
NotificationsWithSubscribers=notifications,
)
In this exercise, you'll build a cost attribution report for a hypothetical production pipeline environment using BigQuery's INFORMATION_SCHEMA.
Setup: You'll need access to a BigQuery project that has run at least some queries in the past 30 days. If you're using a fresh project, run a few queries first.
Step 1: Query the job history
Run the following to get a baseline of your project's query spend:
SELECT
user_email,
COUNT(*) as job_count,
ROUND(SUM(total_bytes_billed) / POW(10, 12) * 6.25, 4) as total_cost_usd,
ROUND(AVG(total_bytes_billed) / POW(10, 9), 2) as avg_gb_per_query
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE
DATE(creation_time) >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
AND job_type = 'QUERY'
AND state = 'DONE'
AND cache_hit = FALSE
GROUP BY user_email
ORDER BY total_cost_usd DESC;
Step 2: Find your most expensive unlabeled queries
-- Queries without pipeline labels are invisible to your attribution model
SELECT
job_id,
user_email,
ROUND(total_bytes_billed / POW(10, 9), 1) as gb_billed,
ROUND(total_bytes_billed / POW(10, 12) * 6.25, 4) as cost_usd,
SUBSTR(query, 0, 100) as query_preview
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE
DATE(creation_time) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
AND job_type = 'QUERY'
AND state = 'DONE'
AND cache_hit = FALSE
AND ARRAY_LENGTH(labels) = 0 -- no labels = no attribution
AND total_bytes_billed > 10 * POW(10, 9) -- more than 10 GB
ORDER BY total_bytes_billed DESC
LIMIT 20;
Step 3: Build your attribution dashboard query
Adapt the query fingerprinting query from the Profiling section and add a column that estimates the potential savings from adding partition filters (measured as the difference between total_bytes_processed and total_bytes_billed — when these diverge significantly, it indicates partition pruning is working partially but not fully).
Step 4: Write a Python script that calls the BigQuery Jobs API (using google-cloud-bigquery) to pull the same data, and writes a cost attribution CSV report organized by pipeline label, falling back to user_email for unlabeled queries.
Step 5: Identify one query in your history that could be optimized, explain why, and write the optimized version. Run both versions with bq query --dry_run to compare estimated bytes processed without incurring cost.
Cloud billing label propagation has latency and scope limitations. In BigQuery, labels applied to a job appear in INFORMATION_SCHEMA.JOBS immediately, but in the billing export (BigQuery billing export to a GCS bucket or BQ dataset), they appear within 1-2 days. On AWS, resource tags need to be activated in Cost Explorer before they're tracked — go to Cost Explorer → Cost Allocation Tags → activate the tag keys you're using.
This usually means your WHERE clause in the incremental block references a non-partition column, so the query engine can't prune partitions. Check the actual query being executed by running dbt compile and examining the compiled SQL. Look for INFORMATION_SCHEMA.JOBS on the compiled query to verify total_bytes_processed is much lower than the full table size.
Skew in Spark almost always originates at a JOIN or GROUP BY. Start with the stage immediately before the skewed stage — that's usually where the shuffle that created the skew happened. Look at the join keys or group-by columns in that stage's query plan and check their value distribution with a countByValue() or groupBy().count(). Keys with dramatically uneven frequency (like NULL or a single dominant value) are almost always the culprit.
This is the cost regression problem, and it's almost always caused by new pipelines being added without cost awareness. The fix is procedural, not technical: make cost estimation part of your pipeline design review checklist. Before any new pipeline is approved for production, it should have an estimated monthly cost attached and a designated budget. The analyze_query_for_cost_risks function above, wired into CI, helps with query-level regressions. For pipeline-level regressions, you need budget alerts (as shown in the create_pipeline_budget_alert example) that fire within the same month the regression appears.
Spot and Preemptible instances save money on average but can increase costs when they're frequently interrupted, because interrupted jobs have to restart and re-read data from scratch. Check your Spot interruption rate in CloudWatch or the EMR console. If interruptions are frequent (more than 10-15% of nodes per hour), the retry overhead may exceed the Spot discount. In this case, use Spot instances only for core/worker nodes, not master nodes, and consider using checkpoint-capable workloads (checkpoint() in Spark or saveAsSequenceFile to HDFS for intermediate results).
You now have a complete framework for understanding and controlling cloud spend in production data workflows. Let's crystallize the key ideas:
Cost attribution requires tagging at the workload level. Service-level billing data tells you almost nothing actionable. Labels on BigQuery jobs, tags on EMR clusters, and structured Spark application names give you the granularity to trace spend to specific pipelines, tasks, and teams.
Profiling drives optimization. Without profiling data — stage-level Spark metrics, INFORMATION_SCHEMA.JOBS analysis, per-query byte counts — optimization is guesswork. Build profiling into your operational routine, not just your incident response.
The five expensive anti-patterns are partition filter bypass, fan-out joins, wildcard scans, over-provisioned clusters, and full recomputation. Each has a specific, testable fix. Implement the fix, measure before and after, and document the savings.
Cost governance prevents regression. A one-time optimization campaign is not a strategy. Automated query analysis in CI, budget alerts with pipeline-level granularity, and cost estimation as part of pipeline design reviews are the institutional mechanisms that make improvements stick.
Storage costs are underestimated. Lifecycle policies, Parquet optimization, and eliminating unnecessary intermediate datasets can match or exceed compute-side savings in mature data platforms.
INFORMATION_SCHEMA attribution query against your production project and find your top five unlabeled, expensive queries. Add labels and set baselines.For deeper learning: explore FinOps Foundation's framework for organizational cost governance, study Apache Iceberg's table format for partition evolution strategies that make incremental processing much cleaner, and look into dbt's exposures and meta features for building cost metadata into your data catalog.
Learning Path: Data Pipeline Fundamentals