07 · Big Data Platforms: Distributed Computing, Orchestration, Scaling

Single-server warehouse:
│
├─ PostgreSQL: ~500 GB, 1-5 second queries
└─ Problem: Can't hold terabytes, can't process petabytes
│
Single-server + external compute:
│
├─ Fact table: 100 GB (on warehouse)
├─ Processing: Apache Spark (100 servers)
└─ Architecture: Data in warehouse, compute scales separately

graph TB
    A["Node 1<br/>Partition A"]
    B["Node 2<br/>Partition B"]
    C["Node 3<br/>Partition C"]
    D["Shuffle &<br/>Sort"]
    E["Reduce<br/>Output"]

    A -->|Map| D
    B -->|Map| D
    C -->|Map| D
    D --> E

# Spark: Distributed data processing
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("warehouse").getOrCreate()

# Read from warehouse
df = spark.read \
    .format("bigquery") \
    .option("table", "project.dataset.fact_sales") \
    .load()

# Transform (lazy — doesn't execute yet)
result = df \
    .filter(df.sale_date >= "2024-01-01") \
    .groupBy("customer_id") \
    .agg(F.sum("amount").alias("total")) \
    .filter(F.col("total") > 1000)

# Execute
result.show()

ETL Pipeline:
   1. Extract from PostgreSQL (1 hour)
   2. Extract from Kafka logs (2 hours)
   3. Wait for both extracts to complete
   4. Merge & transform (1 hour)
   5. Load to Gold layer (30 min)

Problem: If step 3 fails, redeploy all or restart manually?
Solution: Orchestration tool manages dependencies and failure recovery.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryCreateEmptyTableOperator
from datetime import datetime, timedelta

with DAG(
    'daily_warehouse_refresh',
    start_date=datetime(2024, 1, 1),
    schedule_interval='0 2 * * *',  # Daily at 2 AM
    catchup=False,
) as dag:

    extract_task = PythonOperator(
        task_id='extract_from_postgres',
        python_callable=extract_data,
        provide_context=True,
    )

    transform_task = PythonOperator(
        task_id='transform_silver',
        python_callable=transform_data,
        depends_on_past=False,  # Independent retries
    )

    load_task = PythonOperator(
        task_id='load_to_gold',
        python_callable=load_to_gold,
    )

    # Define pipeline
    extract_task >> transform_task >> load_task

-- models/staging/stg_customers.sql

WITH source_data AS (
    SELECT * FROM {{ source('raw', 'customers') }}
),

cleaned_data AS (
    SELECT 
        customer_id,
        LOWER(email) as email,
        CASE 
            WHEN segment IS NULL THEN 'Unknown'
            ELSE segment
        END as segment,
        created_date
    FROM source_data
    WHERE customer_id IS NOT NULL
),

final AS (
    SELECT 
        {{ dbt_utils.surrogate_key(['customer_id']) }} as customer_key,
        *
    FROM cleaned_data
)

SELECT * FROM final

# dbt workflow
dbt run           # Execute all models
dbt test          # Run data quality tests
dbt docs generate # Auto-generate docs
dbt debug         # Check warehouse connection

Spark cluster: 10 servers
├─ Driver (1 server): coordinates
├─ Executor (9 servers): process in parallel
└─ Data is distributed across all executors

Single warehouse: More CPU, More RAM
├─ Snowflake: Scale compute independently
├─ BigQuery: Slots (reserved compute per hour)
└─ Databricks: More workers per cluster

Query 1: SELECT TOP 10 products by revenue (5 seconds, 10 GB scanned)
      └─ Result cached in memory
Query 2: Same query (0.5 seconds, cache hit)
      └─ Saves 10 GB scan and 4.5 seconds

Storage: 100 TB (Cloud Storage)
Compute: Query 1 PB (parallel SQL)
Pricing: $6.25/TB scanned (only pay for what you query)

Storage: S3/GCS (shared, lower cost)
Compute: Snowflake clusters (scale independently)
Pricing: $1-4 per credit (1 credit ≈ 1 server⋅hour)
Scaling: Manual (specify cluster size)

Storage: Delta format (S3/GCS, optimized compression)
Compute: Spark clusters (on EC2/GCP/Azure)
Pricing: $0.30-0.50 per DBU-hour
Scaling: Cluster autoscaling (define min/max size)

graph TB
    A["Data Sources<br/>Apps, APIs, Logs"]
    B["Cloud Storage<br/>S3, GCS, ADLS"]
    C["Ingestion<br/>Fivetran, Stitch,<br/>Custom"]
    D["Orchestration<br/>Airflow, Dagster"]
    E["Transformation<br/>dbt, Spark SQL"]
    F["Warehouse<br/>BigQuery,<br/>Snowflake"]
    G["BI Layer<br/>Looker, Tableau,<br/>Metabase"]

    A --> C
    C --> B
    B --> D
    D --> E
    E --> F
    F --> G