Data Analytics at Scale

Stored on disk:
Row 1: [id=1, name="Alice", age=30, salary=80000, dept="Eng"]
Row 2: [id=2, name="Bob", age=25, salary=60000, dept="Sales"]
Row 3: [id=3, name="Charlie", age=35, salary=90000, dept="Eng"]
...

Query: SELECT AVG(salary) FROM employees WHERE dept = 'Eng';

Problem: Must read ALL columns for ALL rows (name, age, etc.)
even though we only need salary and dept!

Stored on disk:
dept column:   ["Eng", "Sales", "Eng", "Eng", "Sales", ...]
salary column: [80000, 60000, 90000, 75000, 55000, ...]
name column:   ["Alice", "Bob", "Charlie", ...] (not needed)
age column:    [30, 25, 35, ...] (not needed)

Query: SELECT AVG(salary) FROM employees WHERE dept = 'Eng';

Optimization: Only read dept and salary columns!
Compression: "Eng" repeated 1M times compresses to few bytes

┌─────────────────────────────────────────────────┐
│         PRESTO CLUSTER ARCHITECTURE             │
├─────────────────────────────────────────────────┤
│                                                 │
│  ┌──────────────────┐                           │
│  │   Coordinator    │  (Receives query, plans)  │
│  │  - Parse SQL     │                           │
│  │  - Optimize      │                           │
│  │  - Schedule      │                           │
│  └────────┬─────────┘                           │
│           │                                     │
│           │ Distributes tasks                   │
│           ▼                                     │
│  ┌────────┴─────────┬──────────────────┬───┐    │
│  │                  │                  │   │    │
│  ▼                  ▼                  ▼   │    │
│ ┌─────────┐    ┌─────────┐    ┌─────────┐  │    │
│ │Worker 1 │    │Worker 2 │    │Worker N │  │    │
│ │Scan S3  │    │Scan S3  │    │Scan S3  │  │    │
│ │1M rows  │    │1M rows  │    │1M rows  │  │    │
│ └────┬────┘    └────┬────┘    └────┬────┘  │    │
│      │              │              │       │    │
│      └──────────────┼──────────────┘       │    │
│                     │                      │    │
│                     ▼                      │    │
│              ┌──────────────┐              │    │
│              │ Aggregate    │              │    │
│              │ & Return     │              │    │
│              └──────────────┘              │    │
└─────────────────────────────────────────────────┘

-- Define table pointing to S3 Parquet files
CREATE TABLE sales (
  order_id BIGINT,
  customer_id BIGINT,
  product_id BIGINT,
  amount DECIMAL(10,2),
  order_date DATE
)
WITH (
  format = 'PARQUET',
  external_location = 's3://my-data-lake/sales/'
);

-- Query 1 billion rows across 100 workers
SELECT 
  DATE_TRUNC('month', order_date) AS month,
  SUM(amount) AS revenue,
  COUNT(DISTINCT customer_id) AS customers
FROM sales
WHERE order_date >= DATE '2024-01-01'
GROUP BY 1
ORDER BY 1;

-- Join S3 data lake with live PostgreSQL
SELECT 
  c.name,
  c.email,
  SUM(s.amount) AS total_spent
FROM hive.default.sales s           -- S3 / Hive catalog
JOIN postgresql.public.customers c  -- Live Postgres
  ON s.customer_id = c.id
WHERE s.order_date >= DATE '2024-01-01'
GROUP BY c.name, c.email
HAVING SUM(s.amount) > 10000
ORDER BY total_spent DESC
LIMIT 100;

┌──────────────────────────────────────────────────┐
│          SPARK CLUSTER ARCHITECTURE              │
├──────────────────────────────────────────────────┤
│                                                  │
│  ┌─────────────────┐                             │
│  │  Spark Driver   │  (Your code, DAG creation)  │
│  │  - SparkContext │                             │
│  │  - DAG Scheduler│                             │
│  │  - Task         │                             │
│  │    Scheduler    │                             │
│  └────────┬────────┘                             │
│           │                                      │
│           │ Distributes tasks                    │
│           ▼                                      │
│  ┌────────┴─────┬──────────────┬────┐            │
│  │              │              │    │            │
│  ▼              ▼              ▼    │            │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐           │
│ │Executor 1│ │Executor 2│ │Executor N│           │
│ │4 cores   │ │4 cores   │ │4 cores   │           │
│ │16GB RAM  │ │16GB RAM  │ │16GB RAM  │           │
│ │          │ │          │ │          │           │
│ │ Task 1   │ │ Task 3   │ │ Task 5   │           │
│ │ Task 2   │ │ Task 4   │ │ Task 6   │           │
│ └──────────┘ └──────────┘ └──────────┘           │
│      ▲             ▲             ▲               │
│      └─────────────┼─────────────┘               │
│                    │                             │
│             Read from HDFS/S3                    │
└──────────────────────────────────────────────────┘

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, count, date_trunc

# Initialize Spark
spark = SparkSession.builder \
    .appName("Sales Analysis") \
    .config("spark.executor.memory", "16g") \
    .config("spark.executor.cores", "4") \
    .getOrCreate()

# Read Parquet from S3 (billions of rows)
sales_df = spark.read.parquet("s3://data-lake/sales/")

# Perform aggregation (distributed across cluster)
monthly_revenue = sales_df \
    .filter(col("order_date") >= "2024-01-01") \
    .groupBy(date_trunc("month", "order_date").alias("month")) \
    .agg(
        sum("amount").alias("revenue"),
        count("customer_id").alias("orders")
    ) \
    .orderBy("month")

# Execute and show results
monthly_revenue.show()

# Read streaming data from Kafka
stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "clickstream") \
    .load()

# Parse JSON and aggregate in 5-minute windows
from pyspark.sql.functions import window, from_json
from pyspark.sql.types import StructType, StructField, StringType, LongType, TimestampType

schema = StructType([
    StructField("timestamp", TimestampType()),
    StructField("page_url", StringType()),
    StructField("user_id", LongType())
])

clicks_per_page = stream_df \
    .selectExpr("CAST(value AS STRING) as json") \
    .select(from_json("json", schema).alias("data")) \
    .select("data.*") \
    .groupBy(
        window("timestamp", "5 minutes"),
        "page_url"
    ) \
    .count() \
    .orderBy("window", "count", ascending=False)

# Write results to dashboard (continuously)
query = clicks_per_page.writeStream \
    .format("memory") \
    .queryName("top_pages") \
    .outputMode("complete") \
    .start()

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression

# Load training data (millions of rows)
data = spark.read.parquet("s3://ml-data/housing/")

# Prepare features
assembler = VectorAssembler(
    inputCols=["sqft", "bedrooms", "bathrooms", "age"],
    outputCol="features"
)
training_data = assembler.transform(data)

# Split into training and test sets
train_data, test_data = training_data.randomSplit([0.8, 0.2], seed=42)

# Train model (distributed)
lr = LinearRegression(
    featuresCol="features",
    labelCol="price",
    maxIter=10
)
model = lr.fit(train_data)

# Make predictions
predictions = model.transform(test_data)
predictions.select("features", "price", "prediction").show()

┌────────────────────────────────────────────────────────┐
│             DRUID ARCHITECTURE                         │
├────────────────────────────────────────────────────────┤
│                                                        │
│  Ingestion Layer:                                      │
│  ┌──────────┐      ┌──────────┐                        │
│  │  Kafka   │─────▶│ Real-time│  (Indexing Service)    │
│  │ Stream   │      │  Nodes   │  Ingest + Index        │
│  └──────────┘      └─────┬────┘                        │
│                          │                             │
│                          ▼ (persist)                   │
│  Storage Layer:    ┌──────────┐                        │
│  ┌──────────┐      │Deep      │                        │
│  │Historical│◀─────│Storage   │  (S3/HDFS)             │
│  │Nodes     │      │(Segments)│  Column-oriented       │
│  └────┬─────┘      └──────────┘                        │
│       │                                                │
│  Query Layer:                                          │
│       ▼                                                │
│  ┌──────────┐      ┌──────────┐                        │
│  │ Broker   │─────▶│  Query   │                        │
│  │ Nodes    │      │ Results  │  (Merge + Return)      │
│  └────┬─────┘      └──────────┘                        │
│       ▲                                                │
│       │                                                │
│  ┌────┴─────┐                                          │
│  │ Client   │  SELECT ... WHERE timestamp > now()-1h   │
│  └──────────┘                                          │
└────────────────────────────────────────────────────────┘

{
  "type": "kafka",
  "dataSchema": {
    "dataSource": "clickstream",
    "timestampSpec": { "column": "timestamp", "format": "iso" },
    "dimensionsSpec": {
      "dimensions": ["user_id", "page_url", "country", "device"]
    },
    "metricsSpec": [
      { "type": "count", "name": "count" },
      { "type": "longSum", "name": "session_duration", "fieldName": "duration" },
      { "type": "hyperUnique", "name": "unique_users", "fieldName": "user_id" }
    ],
    "granularitySpec": {
      "segmentGranularity": "hour",  // 1 segment per hour
      "queryGranularity": "minute"   // Roll up to minute
    }
  }
}

SELECT 
  TIME_FLOOR(__time, 'PT5M') AS time_bucket,
  country,
  SUM(count) AS page_views,
  APPROX_COUNT_DISTINCT_DS_HLL(unique_users) AS users
FROM clickstream
WHERE __time >= CURRENT_TIMESTAMP - INTERVAL '1' HOUR
GROUP BY 1, 2
ORDER BY 1 DESC, 3 DESC
LIMIT 100;

-- Exact count (slow - scans all events)
SELECT COUNT(DISTINCT user_id) FROM clickstream;
-- Scans: 1 billion rows, Takes: 30 seconds

-- Approximate count (fast - uses HyperLogLog sketch)
SELECT APPROX_COUNT_DISTINCT_DS_HLL(unique_users) 
FROM clickstream;
-- Scans: pre-aggregated segments, Takes: 100ms
-- Accuracy: ±1-2% error