De Cost Optimization

DE Cost Optimization

When to Use

Load this skill when the user needs to:

• Review a cloud bill and identify which pipelines, queries, or clusters drive cost
• Right-size a Spark, Trino, ClickHouse, or Airflow cluster (instance type, memory, core count)
• Reduce query scan costs via partition pruning, Z-order clustering, or file compaction
• Move data to cheaper storage tiers (S3-IA, Glacier, ClickHouse cold volumes)
• Implement spot/preemptible instances for Spark executors or Airflow workers
• Decide whether to materialize a view or a summary table (refresh cost vs query savings)
• Define or enforce a data lifecycle and retention policy
• Set up AWS/GCP budget alerts, FinOps tagging, or per-team cost attribution
• Run a FinOps review and produce a prioritized cost reduction plan

1. Cost Model Overview

Every data engineering cloud bill breaks down into three levers. Attack them in this order — storage is cheapest per GB, but compute at scale dominates the total.

How Each Scales

• Compute scales with cluster lifetime × size. A cluster that is 2× over-provisioned and never auto-scales burns double money even when idle. Idle Spark clusters during off-peak hours are the single most common waste category.
• Storage scales with raw data volume × replication factor × tier. Columnar + compressed formats (Parquet/zstd) reduce this 5-10× vs raw JSON/CSV.
• Data transfer is often invisible until bill arrives. Reading Parquet on S3 from a same-region EMR cluster is free; reading it cross-region costs $0.02/GB and can exceed compute cost on large jobs.

Which to Attack First

Priority 1 — Compute
  ├─ Idle clusters / always-on servers with <20% avg CPU
  ├─ Fixed-size clusters that could be spot + auto-scale
  └─ Over-provisioned executor memory

Priority 2 — Storage
  ├─ Raw JSON/CSV on S3 Standard (should be Parquet + tiered)
  ├─ Small files (<10 MB) causing excess LIST/GET API calls
  └─ Data past retention policy still sitting in S3 Standard

Priority 3 — Data Transfer
  ├─ Queries on cross-region tables
  └─ Shuffles writing to S3 between EMR and Glue in different AZs

2. Query Cost Analysis

2.1 Trino / Presto

Reading EXPLAIN (TYPE DISTRIBUTED)

EXPLAIN (TYPE DISTRIBUTED)
SELECT customer_id, SUM(amount) AS revenue
FROM orders
WHERE order_date >= DATE '2024-01-01'
  AND region = 'US'
GROUP BY customer_id;

Key tokens to look for in the output:

Partition filter pushdown is confirmed when the TableScan node shows partitionedBy columns in its predicate. If your WHERE order_date = ... clause does NOT appear in the TableScan predicate, it means the filter is not pushed down — you are scanning the whole table and filtering in memory.

Finding Expensive Queries via system tables

-- Top 20 most expensive queries in the last 24 hours (Trino)
SELECT
    query_id,
    LEFT(query, 120)                                          AS query_snippet,
    user,
    source,
    state,
    CAST(read_data_size AS VARCHAR)                           AS scanned,
    -- Estimate cost at $5/TB scanned (Athena pricing proxy)
    ROUND(read_data_size / 1e12 * 5.0, 4)                    AS est_cost_usd,
    DATE_DIFF('second', created, "end")                       AS wall_sec,
    peak_memory_bytes / 1e9                                   AS peak_memory_gb,
    cpu_time_ms / 1000.0                                      AS cpu_sec
FROM system.runtime.query_history
WHERE created >= NOW() - INTERVAL '24' HOUR
  AND state = 'FINISHED'
ORDER BY read_data_size DESC
NULLS LAST
LIMIT 20;

-- Aggregate cost per source (team / application) over last 7 days
SELECT
    source,
    COUNT(*)                                                  AS query_count,
    ROUND(SUM(read_data_size) / 1e12, 3)                     AS total_tb_scanned,
    ROUND(SUM(read_data_size) / 1e12 * 5.0, 2)               AS est_total_cost_usd,
    ROUND(AVG(DATE_DIFF('second', created, "end")), 1)        AS avg_wall_sec
FROM system.runtime.query_history
WHERE created >= NOW() - INTERVAL '7' DAY
  AND state = 'FINISHED'
GROUP BY source
ORDER BY total_tb_scanned DESC;

File Size Optimization

• Too-small files (<10 MB): excessive S3 LIST + GET overhead; Trino opens one reader thread per file, so 10,000 files × overhead > reading one 1 GB file.
• Too-large files (>1 GB): cannot be split across workers in parallel; one task processes the whole file.
• Sweet spot: 128–512 MB per file after compression.

-- Find partitions with small-file problem (Iceberg metadata)
SELECT
    partition,
    COUNT(*)           AS file_count,
    AVG(file_size_in_bytes) / 1e6 AS avg_file_mb,
    SUM(file_size_in_bytes) / 1e9 AS total_gb
FROM "catalog"."schema"."table$files"
GROUP BY partition
HAVING COUNT(*) > 100 OR AVG(file_size_in_bytes) < 10e6
ORDER BY file_count DESC
LIMIT 30;

2.2 Spark (EMR / Databricks / Self-Hosted)

Spark UI Cost Signals

Navigate: Jobs → select expensive job → Stages → sort by Duration descending.

Key indicators:

• Shuffle Read/Write size: A stage with 500 GB shuffle write followed by a stage with 500 GB shuffle read means you are moving 500 GB over the network. Each GB shuffled costs executor CPU time + network I/O + potential spill to disk.
• GC time % > 10%: executors spending >10% in garbage collection → memory is too low for the data size; increase spark.executor.memory or reduce partition size.
• Task skew: longest task 10× median → one partition is much larger than others (data skew). Fix with salting or AQE.
• Spill (Memory) / Spill (Disk): non-zero spill means tasks ran out of memory and wrote to disk → slower and wastes I/O.

EXPLAIN COST

spark.sql("""
    EXPLAIN COST
    SELECT customer_id, SUM(amount) AS revenue
    FROM orders
    WHERE order_date >= '2024-01-01'
    GROUP BY customer_id
""").show(truncate=False)

Look for Statistics(sizeInBytes=..., rowCount=...). If estimates are wildly wrong, run ANALYZE TABLE orders COMPUTE STATISTICS FOR ALL COLUMNS to refresh them — bad statistics cause the planner to choose wrong join strategies (BroadcastHashJoin vs SortMergeJoin).

Executor Sizing Formula

spark.executor.memory  = 4–8 GB per executor core (memory-heavy ETL)
spark.executor.cores   = 4–5 cores per executor (limits context switches)
Total executors        = (Total vCPUs allocated) / spark.executor.cores

Example for r5.2xlarge (8 vCPU, 64 GB):
  spark.executor.cores   = 4
  spark.executor.memory  = 12g   (leaving 4g for OS + overhead)
  spark.executor.memoryOverhead = 2g
  Executors per node     = 2

Dynamic Allocation vs Fixed: Cost Comparison

# Fixed cluster: 10 executors running for 8 hours
fixed_cost = 10 * 8 * instance_cost_per_hour   # e.g. 10 * 8 * $0.50 = $40

# Dynamic allocation: average 3 executors over 8 hours
dynamic_cost = 3 * 8 * instance_cost_per_hour  # $12  → 70% savings

# Enable dynamic allocation
spark = (
    SparkSession.builder
    .config("spark.dynamicAllocation.enabled", "true")
    .config("spark.dynamicAllocation.minExecutors", "1")
    .config("spark.dynamicAllocation.maxExecutors", "50")
    .config("spark.dynamicAllocation.initialExecutors", "2")
    .config("spark.dynamicAllocation.executorIdleTimeout", "120s")
    .config("spark.dynamicAllocation.schedulerBacklogTimeout", "30s")
    .config("spark.shuffle.service.enabled", "true")   # required for dynamic alloc
    .getOrCreate()
)

Reading EMR Utilization from CloudWatch

import boto3
from datetime import datetime, timedelta

def get_cluster_cpu_utilization(cluster_id: str, days: int = 7) -> float:
    """Return average YARNMemoryAvailablePercentage over last N days."""
    cw = boto3.client("cloudwatch", region_name="us-east-1")
    resp = cw.get_metric_statistics(
        Namespace="AWS/ElasticMapReduce",
        MetricName="YARNMemoryAvailablePercentage",
        Dimensions=[{"Name": "JobFlowId", "Value": cluster_id}],
        StartTime=datetime.utcnow() - timedelta(days=days),
        EndTime=datetime.utcnow(),
        Period=3600,
        Statistics=["Average"],
    )
    datapoints = resp["Datapoints"]
    if not datapoints:
        return 0.0
    avg = sum(d["Average"] for d in datapoints) / len(datapoints)
    return round(100 - avg, 1)   # invert: available% → used%

2.3 ClickHouse

Identifying Expensive Queries

-- Top 20 most expensive queries by bytes read, last 24 hours
SELECT
    query_id,
    user,
    LEFT(query, 100)                                       AS query_snippet,
    read_bytes / 1e9                                       AS read_gb,
    read_rows,
    memory_usage / 1e9                                     AS memory_gb,
    query_duration_ms / 1000.0                             AS duration_sec,
    -- Mark queries that could benefit from projection
    result_rows < read_rows / 100                          AS high_reduction_ratio
FROM system.query_log
WHERE event_time >= NOW() - INTERVAL 1 DAY
  AND type = 'QueryFinish'
  AND query_kind = 'Select'
ORDER BY read_bytes DESC
LIMIT 20;

-- Aggregate read cost by user over last 7 days
SELECT
    user,
    COUNT(*)                        AS query_count,
    ROUND(SUM(read_bytes) / 1e12, 3) AS total_tb_read,
    ROUND(AVG(query_duration_ms) / 1000, 1) AS avg_sec,
    ROUND(SUM(memory_usage) / 1e12, 3)       AS total_memory_tb
FROM system.query_log
WHERE event_time >= NOW() - INTERVAL 7 DAY
  AND type = 'QueryFinish'
  AND query_kind = 'Select'
GROUP BY user
ORDER BY total_tb_read DESC;

Verifying Projection Usage

-- Check whether a projection is being used for a specific query
-- Run EXPLAIN SELECT ... and look for "ReadFromMergeTree (projection: <name>)"
EXPLAIN
SELECT region, SUM(amount)
FROM orders
WHERE order_date >= '2024-01-01'
GROUP BY region;

-- List all projections in a table
SELECT
    name,
    part_name,
    rows,
    data_compressed_bytes / 1e6 AS compressed_mb
FROM system.projection_parts
WHERE table = 'orders'
ORDER BY data_compressed_bytes DESC;

Cold vs Hot Data Cost Model (TTL MOVE)

-- Configure tiered storage: hot data on NVMe, cold data on S3
ALTER TABLE events
    MODIFY TTL
        event_date + INTERVAL 30 DAY TO VOLUME 'cold_s3',  -- move to S3 after 30d
        event_date + INTERVAL 365 DAY DELETE;               -- delete after 1 year

-- Estimate cost: if S3 = $0.023/GB-month vs NVMe = $0.10/GB-month
-- Moving 10 TB of 31-90 day data from NVMe to S3 saves ~$0.077/GB-month = $770/month

2.4 BigQuery

Finding Expensive Queries

-- Top 20 most bytes-billed queries, last 7 days (BigQuery INFORMATION_SCHEMA)
SELECT
    job_id,
    user_email,
    LEFT(query, 120)                                         AS query_snippet,
    total_bytes_billed / POW(1024, 4)                        AS tb_billed,
    -- On-demand: $6.25/TB
    ROUND(total_bytes_billed / POW(1024, 4) * 6.25, 4)      AS est_cost_usd,
    total_slot_ms / 1000.0 / 3600.0                          AS slot_hours,
    TIMESTAMP_DIFF(end_time, start_time, SECOND)             AS duration_sec,
    creation_time
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 20;

-- Cost by user over last 30 days
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 est_cost_usd,
    ROUND(SUM(total_slot_ms) / 1e6, 1)                       AS total_slot_ksec
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
  AND job_type = 'QUERY'
GROUP BY user_email
ORDER BY est_cost_usd DESC;

Partition Expiration and Clustering Verification

-- Check whether clustering columns are being used (table metadata)
SELECT
    table_name,
    clustering_fields,
    partition_expiration_days,
    require_partition_filter   -- TRUE = force WHERE on partition column
FROM `project.dataset`.INFORMATION_SCHEMA.TABLES
WHERE table_name = 'orders';

-- Identify tables without partition filter requirement (cost risk)
SELECT table_name, row_count, size_bytes / 1e9 AS size_gb
FROM `project.dataset`.INFORMATION_SCHEMA.TABLE_STORAGE
WHERE total_partitions > 10
  AND table_name NOT IN (
      SELECT table_name FROM `project.dataset`.INFORMATION_SCHEMA.TABLES
      WHERE require_partition_filter = TRUE
  )
ORDER BY size_bytes DESC;

3. Compute Right-Sizing

Memory-to-vCPU Ratios for DE Workloads

Identifying Over-Provisioned Clusters

A cluster with average CPU utilization below 20% over 7 days is over-provisioned. Correct action is to either downsize or enable auto-scaling.

import boto3
from datetime import datetime, timedelta

def is_over_provisioned(cluster_id: str, threshold_pct: float = 20.0) -> bool:
    """Return True if EMR cluster avg CPU < threshold over last 7 days."""
    cw = boto3.client("cloudwatch", region_name="us-east-1")
    resp = cw.get_metric_statistics(
        Namespace="AWS/ElasticMapReduce",
        MetricName="CoreNodesPending",
        Dimensions=[{"Name": "JobFlowId", "Value": cluster_id}],
        StartTime=datetime.utcnow() - timedelta(days=7),
        EndTime=datetime.utcnow(),
        Period=86400,
        Statistics=["Average"],
    )
    # Also check YARNMemoryAvailablePercentage — high available% = over-provisioned
    resp2 = cw.get_metric_statistics(
        Namespace="AWS/ElasticMapReduce",
        MetricName="YARNMemoryAvailablePercentage",
        Dimensions=[{"Name": "JobFlowId", "Value": cluster_id}],
        StartTime=datetime.utcnow() - timedelta(days=7),
        EndTime=datetime.utcnow(),
        Period=3600,
        Statistics=["Average"],
    )
    datapoints = resp2["Datapoints"]
    if not datapoints:
        return False
    avg_available = sum(d["Average"] for d in datapoints) / len(datapoints)
    avg_used = 100 - avg_available
    return avg_used < threshold_pct

Kubernetes Resource Requests vs Limits

Every pod that requests 4 CPU but uses 0.5 CPU reserves 3.5 CPU that the cluster cannot schedule elsewhere — you pay for reserved-but-idle capacity.

# Over-provisioned (anti-pattern): requests 4 CPU but job uses ~0.5 CPU
resources:
  requests:
    cpu: "4"
    memory: "16Gi"
  limits:
    cpu: "8"
    memory: "32Gi"

# Right-sized (based on 7-day P95 from Prometheus):
resources:
  requests:
    cpu: "1"        # P50 usage + 20% headroom
    memory: "4Gi"
  limits:
    cpu: "2"        # burst headroom
    memory: "6Gi"

# Find over-provisioned pods: requested >> actual usage (requires metrics-server)
kubectl top pods -n data-engineering --sort-by=cpu | head -30

# Goldilocks can auto-recommend right-sized requests/limits
helm install goldilocks fairwinds-stable/goldilocks
kubectl label namespace data-engineering goldilocks.fairwinds.com/enabled=true

Auto-Scaling Rules

# Kubernetes HPA for Airflow workers — scale at 70% CPU
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: airflow-worker-hpa
  namespace: airflow
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: airflow-worker
  minReplicas: 1
  maxReplicas: 20
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 600   # wait 10 min before scaling down
      policies:
        - type: Pods
          value: 2
          periodSeconds: 120
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
        - type: Percent
          value: 100
          periodSeconds: 30

4. Spot / Preemptible Instances Strategy

Which Workloads Are Spot-Safe

Checkpointing Requirement for Spot-Safe Spark Jobs

# A Spark job is spot-safe ONLY when:
# 1. Checkpoint location is on persistent storage (S3, HDFS)
# 2. Task max failures is high enough to survive interruptions
# 3. Job output is idempotent (partition overwrite or MERGE, not append)

spark = (
    SparkSession.builder
    .appName("spot_safe_etl")
    .config("spark.task.maxFailures", "8")          # allow more task retries
    .config("spark.stage.maxConsecutiveAttempts", "8")
    .config("spark.speculation", "false")            # disable speculation on spot
    .config("spark.checkpoint.compress", "true")
    .getOrCreate()
)

sc = spark.sparkContext
sc.setCheckpointDir("s3://my-checkpoints/spark/spot_safe_etl/")

AWS Spot Interruption Handler Pattern

# Install on each Spot instance via EC2 user-data or init action
# Polls IMDS for interruption notice every 5 seconds; on SIGTERM gracefully drains

import threading
import signal
import urllib.request
import logging

logger = logging.getLogger("spot_handler")

_INTERRUPTED = threading.Event()

def _poll_interruption_notice(interval: float = 5.0):
    """AWS provides a 2-minute warning via IMDS before spot reclamation."""
    while not _INTERRUPTED.is_set():
        try:
            url = "http://169.254.169.254/latest/meta-data/spot/interruption-action"
            with urllib.request.urlopen(url, timeout=1) as r:
                action = r.read().decode()
                logger.warning("spot_interruption_notice", extra={"action": action})
                _INTERRUPTED.set()
                # Graceful shutdown: flush buffers, checkpoint, notify Spark
                _graceful_shutdown()
        except Exception:
            pass  # 404 = no interruption notice yet
        _INTERRUPTED.wait(timeout=interval)


def _graceful_shutdown():
    """Flush write buffers and send SIGTERM to Spark driver."""
    import os
    logger.info("initiating_graceful_shutdown")
    # Allow Spark tasks to finish current micro-batch
    os.kill(os.getpid(), signal.SIGTERM)


# Start poller thread in each executor / worker process
threading.Thread(target=_poll_interruption_notice, daemon=True).start()

Multi-AZ Spot Fleet with Fallback

{
  "SpotFleetRequestConfig": {
    "AllocationStrategy": "diversified",
    "TargetCapacity": 20,
    "IamFleetRole": "arn:aws:iam::123456789:role/SpotFleetRole",
    "LaunchSpecifications": [
      {
        "InstanceType": "r5.2xlarge",
        "SubnetId": "subnet-us-east-1a",
        "WeightedCapacity": 4
      },
      {
        "InstanceType": "r5.2xlarge",
        "SubnetId": "subnet-us-east-1b",
        "WeightedCapacity": 4
      },
      {
        "InstanceType": "r5a.2xlarge",
        "SubnetId": "subnet-us-east-1a",
        "WeightedCapacity": 4
      },
      {
        "InstanceType": "m5.4xlarge",
        "SubnetId": "subnet-us-east-1c",
        "WeightedCapacity": 4
      }
    ],
    "OnDemandTargetCapacity": 2,
    "SpotMaintenanceStrategies": {
      "CapacityRebalance": {"ReplacementStrategy": "priceCapacityOptimized"}
    }
  }
}

Savings: Spot instances typically cost 60–90% less than on-demand. Moving 80% of Spark executors to spot with 2 on-demand fallback nodes cuts executor cost by ~70%.

5. Storage Optimization

5.1 Partitioning and File Layout

Partition Key Selection Rules

Always partition on the column that appears in WHERE clauses of 80%+ of queries. event_date is the canonical choice for time-series data.

File Compaction (Delta Lake)

-- Compact small files in a partition, targeting 256 MB files
OPTIMIZE orders
WHERE event_date >= '2024-01-01'
ZORDER BY (customer_id, region);

-- Schedule as off-peak Airflow task

# PySpark: manual compaction for non-Delta Parquet
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Read all small files in partition, write back as fewer large files
df = spark.read.parquet("s3://datalake/orders/event_date=2024-01-01/")

(
    df.repartition(8)           # 8 files × ~256MB = ~2GB partition
    .write
    .mode("overwrite")
    .parquet("s3://datalake/orders/event_date=2024-01-01/")
)

Compaction trigger thresholds:

• Run compaction when average file size drops below 32 MB in a partition
• Or when file count in a partition exceeds 500 files

Z-Order / Clustering for High-Cardinality Filter Columns

Z-ordering is for columns you filter on that have too high a cardinality to partition by. It co-locates rows with similar values in the same files, enabling data-skipping.

-- Delta Lake: Z-order on customer_id and product_category
-- (these are NOT partition columns — they are high-cardinality filter columns)
OPTIMIZE orders
ZORDER BY (customer_id, product_category);

-- Iceberg: sort order for clustering (write-time sort, not post-hoc)
ALTER TABLE orders
SET WRITE DISTRIBUTED BY PARTITION LOCALLY ORDERED BY (customer_id, product_category);

-- Verify data-skipping effect: check files_scanned before and after
-- Before Z-ORDER: EXPLAIN shows 10,000 files scanned for customer_id = 12345
-- After Z-ORDER:  EXPLAIN shows 12 files scanned for customer_id = 12345

Z-ORDER Cost vs Benefit Analysis

OPTIMIZE + ZORDER cost:
  - Reads all files in target partition(s): e.g. 500 GB read
  - Rewrites all files sorted: 500 GB write
  - S3 cost at $0.023/GB: (500 + 500) × $0.023 = $23 one-time

Daily savings (if top query runs 100× per day, scanning 100 GB without Z-ORDER):
  - Without Z-ORDER: 100 queries × 100 GB × $5/TB × 0.001 = $0.05/day scan cost (Athena proxy)
  - With Z-ORDER (99% skip rate): 100 queries × 1 GB × ... = $0.0005/day
  - Daily savings ≈ $0.05 - $0.0005 = ~$0.0495/day
  - Break-even: $23 / $0.0495 ≈ 465 days (not worth it at this scale)
  
Rule of thumb: Z-ORDER pays off when the workload scans > 1 TB/day on a column
and achieves > 80% skip rate, OR when compute cost (not scan cost) dominates.

Iceberg Table Maintenance

-- Expire old snapshots (keep last 7 days of history)
CALL catalog.system.expire_snapshots(
    table => 'schema.orders',
    older_than => TIMESTAMP '2024-01-01 00:00:00',
    retain_last => 5
);

-- Remove orphan files (unreferenced files left by failed writes)
CALL catalog.system.remove_orphan_files(
    table => 'schema.orders',
    older_than => TIMESTAMP '2024-01-01 00:00:00'
);

-- Schedule both as weekly Airflow maintenance tasks

5.2 Storage Tiers

S3 Tier Selection Guide

S3 Intelligent-Tiering break-even: worthwhile when >30% of objects are not accessed for >30 days. Enable on the raw/bronze bucket where access patterns are unpredictable.

S3 Lifecycle Rules (JSON)

{
  "Rules": [
    {
      "ID": "bronze-to-ia-after-30d",
      "Status": "Enabled",
      "Filter": {"Prefix": "bronze/"},
      "Transitions": [
        {"Days": 30,  "StorageClass": "STANDARD_IA"},
        {"Days": 90,  "StorageClass": "GLACIER_IR"},
        {"Days": 365, "StorageClass": "DEEP_ARCHIVE"}
      ]
    },
    {
      "ID": "silver-to-ia-after-90d",
      "Status": "Enabled",
      "Filter": {"Prefix": "silver/"},
      "Transitions": [
        {"Days": 90,  "StorageClass": "STANDARD_IA"},
        {"Days": 730, "StorageClass": "GLACIER_IR"}
      ]
    },
    {
      "ID": "gold-keep-standard",
      "Status": "Enabled",
      "Filter": {"Prefix": "gold/"},
      "Transitions": [
        {"Days": 1825, "StorageClass": "STANDARD_IA"}
      ]
    },
    {
      "ID": "tmp-delete-after-7d",
      "Status": "Enabled",
      "Filter": {"Prefix": "tmp/"},
      "Expiration": {"Days": 7}
    }
  ]
}

Apply with Terraform:

resource "aws_s3_bucket_lifecycle_configuration" "datalake_lifecycle" {
  bucket = aws_s3_bucket.datalake.id

  rule {
    id     = "bronze-tiering"
    status = "Enabled"
    filter { prefix = "bronze/" }

    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }
    transition {
      days          = 90
      storage_class = "GLACIER_IR"
    }
    transition {
      days          = 365
      storage_class = "DEEP_ARCHIVE"
    }
  }

  rule {
    id     = "tmp-cleanup"
    status = "Enabled"
    filter { prefix = "tmp/" }
    expiration { days = 7 }
  }
}

5.3 Format and Compression Comparison

Compression Codec Decision

# Write Parquet with zstd compression (best ratio for archival)
df.write \
    .option("compression", "zstd") \
    .parquet("s3://archive/orders/")

# Write Parquet with snappy (default — best for hot/interactive)
df.write \
    .option("compression", "snappy") \
    .parquet("s3://datalake/orders/")

Column Pruning Impact

Reading only the columns needed reduces scan cost proportionally to column width.

# Expensive: reads all 200 columns
df = spark.read.parquet("s3://datalake/events/")

# Cheap: reads only 5 of 200 columns — 97.5% scan reduction
df = spark.read.parquet("s3://datalake/events/").select(
    "event_id", "user_id", "event_type", "event_time", "amount"
)

In Trino, SELECT * from a wide Parquet table is the single easiest cost win to avoid — always select only needed columns.

6. Materialized View Economics

When to Materialize

Apply the 10-10-15 rule: materialize when the query:

1. Runs more than 10 times per day
2. Takes more than 10 seconds (or scans > 10 GB) each run
3. Has a stale-data tolerance of more than 15 minutes

Refresh Cost vs Query Savings — Break-Even Formula

Daily query cost (without MV) = query_count_per_day × scan_GB × $/GB
Daily refresh cost (with MV)  = refreshes_per_day × full_table_scan_GB × $/GB

Break-even condition:
  refreshes_per_day × full_table_scan_GB < query_count_per_day × scan_GB × hit_rate

Example:
  Query runs 200×/day, scans 50 GB each: cost = 200 × 50 × $0.005 = $50/day
  MV refresh every 15 min (96 refreshes/day), full scan = 200 GB:
    cost = 96 × 200 × $0.005 = $96/day   → NOT WORTH IT

  MV refresh every 1h (24 refreshes/day), incremental scan = 10 GB:
    cost = 24 × 10 × $0.005 = $1.20/day  → SAVINGS = $50 - $1.20 = $48.80/day ✓

ClickHouse Materialized Views: Zero-Cost Streaming Aggregation

ClickHouse MVs process only new rows as they are inserted — no periodic full refresh cost.

-- Source table
CREATE TABLE orders_raw (
    order_id  UInt64,
    region    LowCardinality(String),
    amount    Float64,
    created_at DateTime
) ENGINE = MergeTree()
ORDER BY (created_at, region);

-- AggregatingMergeTree target for the MV
CREATE TABLE orders_hourly_agg (
    hour      DateTime,
    region    LowCardinality(String),
    order_count AggregateFunction(count),
    total_amount AggregateFunction(sum, Float64)
) ENGINE = AggregatingMergeTree()
ORDER BY (hour, region);

-- Materialized view: fires on every INSERT to orders_raw, zero extra cost
CREATE MATERIALIZED VIEW orders_hourly_mv TO orders_hourly_agg AS
SELECT
    toStartOfHour(created_at)   AS hour,
    region,
    countState()                AS order_count,
    sumState(amount)            AS total_amount
FROM orders_raw
GROUP BY hour, region;

-- Query using -Merge combinator to finalize aggregates
SELECT
    hour,
    region,
    countMerge(order_count)     AS orders,
    sumMerge(total_amount)      AS revenue
FROM orders_hourly_agg
WHERE hour >= NOW() - INTERVAL 7 DAY
GROUP BY hour, region
ORDER BY hour, region;

Trino / Spark: CTAS-Based Materialized Table + Scheduled Refresh

-- Step 1: create materialized summary table (run once)
CREATE TABLE analytics.orders_daily_summary
WITH (
    format = 'PARQUET',
    partitioned_by = ARRAY['summary_date']
)
AS
SELECT
    DATE(order_time)               AS summary_date,
    region,
    COUNT(*)                       AS order_count,
    SUM(amount)                    AS revenue,
    COUNT(DISTINCT customer_id)    AS unique_customers
FROM raw.orders
GROUP BY DATE(order_time), region;

# Airflow DAG: incremental refresh of materialized summary (runs hourly)
from airflow.sdk import dag, task
import pendulum

@dag(
    dag_id="refresh_orders_daily_summary",
    schedule="@hourly",
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
)
def refresh_summary():

    @task
    def refresh_last_2_days():
        from trino import dbapi
        conn = dbapi.connect(host="trino", port=8080, user="etl")
        cur = conn.cursor()
        # Delete and re-insert only the last 2 days (incremental, not full scan)
        cur.execute("""
            DELETE FROM analytics.orders_daily_summary
            WHERE summary_date >= CURRENT_DATE - INTERVAL '2' DAY
        """)
        cur.execute("""
            INSERT INTO analytics.orders_daily_summary
            SELECT
                DATE(order_time),
                region,
                COUNT(*),
                SUM(amount),
                COUNT(DISTINCT customer_id)
            FROM raw.orders
            WHERE DATE(order_time) >= CURRENT_DATE - INTERVAL '2' DAY
            GROUP BY DATE(order_time), region
        """)
        conn.commit()

    refresh_last_2_days()

refresh_summary()

7. Data Lifecycle and Retention Policies

Retention Policy Matrix

Implementing Retention via Delta VACUUM

-- Delete data older than retention threshold (Delta Lake)
-- VACUUM removes files no longer referenced by the table; default retention = 7 days
VACUUM orders RETAIN 168 HOURS;  -- keep 7 days of time-travel; delete older files

-- Set table-level retention property
ALTER TABLE orders SET TBLPROPERTIES (
    'delta.deletedFileRetentionDuration' = 'interval 7 days',
    'delta.logRetentionDuration'         = 'interval 30 days'
);

# Airflow task: weekly VACUUM for all Delta tables
from airflow.sdk import dag, task
import pendulum

@dag(
    dag_id="delta_maintenance",
    schedule="0 2 * * 0",   # every Sunday 02:00 UTC
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
)
def delta_maintenance():

    @task
    def vacuum_tables():
        from pyspark.sql import SparkSession
        spark = SparkSession.builder.getOrCreate()
        tables = [
            "datalake.bronze.orders",
            "datalake.bronze.events",
            "datalake.silver.orders_clean",
        ]
        for table in tables:
            spark.sql(f"VACUUM {table} RETAIN 168 HOURS")

    vacuum_tables()

delta_maintenance()

Implementing Retention via Iceberg expire_snapshots

-- Airflow-triggered Trino call
CALL catalog.system.expire_snapshots(
    table        => 'bronze.orders',
    older_than   => CURRENT_TIMESTAMP - INTERVAL '30' DAY,
    retain_last  => 3,
    max_concurrent_deletes => 10
);

CALL catalog.system.remove_orphan_files(
    table      => 'bronze.orders',
    older_than => CURRENT_TIMESTAMP - INTERVAL '7' DAY
);

8. Cost Attribution and Tagging

AWS Resource Tagging Strategy

All data engineering resources must carry these tags consistently:

Tagging EMR Clusters from Airflow

from airflow.providers.amazon.aws.operators.emr import EmrCreateJobFlowOperator

emr_task = EmrCreateJobFlowOperator(
    task_id="create_emr_cluster",
    aws_conn_id="aws_default",
    job_flow_overrides={
        "Name": "orders-daily-load",
        "ReleaseLabel": "emr-7.0.0",
        "Instances": {
            "MasterInstanceType": "m5.xlarge",
            "SlaveInstanceType":  "r5.2xlarge",
            "InstanceCount": 5,
        },
        "Tags": [
            {"Key": "team",         "Value": "analytics"},
            {"Key": "project",      "Value": "orders-etl"},
            {"Key": "env",          "Value": "prod"},
            {"Key": "dag_id",       "Value": "orders_daily_load"},
            {"Key": "cost_center",  "Value": "CC-1234"},
        ],
        "Applications": [{"Name": "Spark"}],
        "JobFlowRole": "EMR_EC2_DefaultRole",
        "ServiceRole": "EMR_DefaultRole",
    },
)

Spark Job Tagging

spark = (
    SparkSession.builder
    .appName("orders_daily_load")
    .config("spark.yarn.tags", "team=analytics,project=orders-etl,env=prod")
    # On K8s: labels propagate to pod metadata → show up in K8s cost tools
    .config("spark.kubernetes.driver.label.team",        "analytics")
    .config("spark.kubernetes.driver.label.project",     "orders-etl")
    .config("spark.kubernetes.executor.label.team",      "analytics")
    .config("spark.kubernetes.executor.label.project",   "orders-etl")
    .getOrCreate()
)

Trino Query Tagging

-- Prepend a comment tag to every Trino query from your application
-- Trino records source + query text; the tag becomes searchable in system.runtime.query_history

-- In application code:
SELECT /* @tag team=analytics project=orders-etl dag_id=orders_daily */ 
    customer_id, SUM(amount)
FROM orders
WHERE order_date = CURRENT_DATE
GROUP BY customer_id;

# Python: inject tags into every query via connection wrapper
import trino

class TaggedTrinoConnection:
    def __init__(self, host, port, user, tags: dict):
        self._conn = trino.dbapi.connect(
            host=host, port=port, user=user,
            source=",".join(f"{k}={v}" for k, v in tags.items()),
        )

    def execute(self, sql: str):
        cur = self._conn.cursor()
        tag_comment = " ".join(f"@{k}={v}" for k, v in self._tags.items())
        cur.execute(f"/* {tag_comment} */ {sql}")
        return cur

Cost per DAG Run Calculation

# Post-run cost attribution: query EMR billing + query bytes
import boto3
from datetime import datetime

def calculate_dag_run_cost(
    cluster_id: str,
    start_time: datetime,
    end_time: datetime,
    instance_type: str = "r5.2xlarge",
    instance_count: int = 5,
) -> dict:
    # On-demand price (look up from AWS Price List API or hardcode)
    INSTANCE_PRICES = {"r5.2xlarge": 0.504, "m5.xlarge": 0.192}
    hourly_rate = INSTANCE_PRICES.get(instance_type, 0.5)
    
    duration_hours = (end_time - start_time).total_seconds() / 3600
    compute_cost = duration_hours * instance_count * hourly_rate
    
    # Spot savings (if 80% executors on spot at 30% of on-demand)
    spot_savings = compute_cost * 0.8 * 0.7
    effective_cost = compute_cost - spot_savings
    
    return {
        "dag_id": "orders_daily_load",
        "cluster_id": cluster_id,
        "duration_hours": round(duration_hours, 3),
        "on_demand_cost_usd": round(compute_cost, 4),
        "effective_cost_usd": round(effective_cost, 4),
        "spot_savings_usd": round(spot_savings, 4),
    }

9. Budget Alerts and FinOps Metrics

AWS Budgets Setup (Terraform)

# Terraform: budget alert per team tag, alert at 80% and 100% of

…(truncated)