Traditional Hadoop (MapReduce):
  Read disk → Process → Write disk → Read disk → Process → Write disk
  Slow because: disk I/O between stages

Spark:
  Read disk → Process (in-memory) → Process → Process → Write disk
  Fast because: intermediate results stay in memory

1. Create RDD/DataFrame
2. Apply transformations (lazy - not executed yet)
   - map, filter, flatMap, join, groupBy
3. Call action (triggers execution)
   - collect(), count(), save(), show()
4. Driver creates DAG (directed acyclic graph)
5. Scheduler submits tasks to executors
6. Results collected back to driver

from pyspark.sql import SparkSession

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

# Read data (lazy)
df = spark.read.csv("data.csv", header=True)

# Transformations (lazy - not executed)
df_filtered = df.filter(df['age'] > 25)
df_grouped = df_filtered.groupBy('country').count()

# Action (executes all transformations)
df_grouped.show()  # Only here does actual computation happen

# Output:
# +-------+-----+
# |country|count|
# +-------+-----+
# |USA    |1000 |
# |UK     |500  |
# +-------+-----+

# Default: auto-partitioned based on data size
df = spark.read.csv("1GB_file.csv")  # Maybe 16 partitions

# Explicit repartitioning
df_repartitioned = df.repartition(32)  # Now 32 partitions
df_repartitioned.show()

# Check partition count
print(df.rdd.getNumPartitions())  # Output: 32

# This causes shuffle (expensive)
df.groupBy('country').count()
# Data must move to same executor for same country

# Minimize shuffles in real jobs

Narrow: One partition output depends on one input partition
- map, filter, select
- No data movement needed

Wide: One partition output depends on multiple input partitions  
- groupBy, join, shuffle
- Data must move across network
- Expensive!

# Narrow (fast)
df.filter(df['age'] > 25)  # Each partition filters independently

# Wide (slow - shuffle)
df.groupBy('country').count()  # Data from all partitions must move

# Too few partitions: underutilization
df.repartition(2)  # Only 2 executors working, others idle

# Too many partitions: overhead
df.repartition(10000)  # Task creation overhead > benefit

# Sweet spot: partition size 128MB
# For 100GB file: 100GB / 128MB = ~800 partitions

# Driver memory: holds results
spark = SparkSession.builder \
    .config("spark.driver.memory", "8g") \
    .appName("app") \
    .getOrCreate()

# Executor memory: processing
spark.conf.set("spark.executor.memory", "16g")

# Too much: GC pauses
# Too little: OOM errors

# Without cache:
df.groupBy('id').count().show()  # Recompute from source
df.groupBy('id').sum('amount').show()  # Recompute again

# With cache:
df.cache()  # Store in memory after first use
df.groupBy('id').count().show()  # Compute once, cache
df.groupBy('id').sum('amount').show()  # Use cached version (fast!)

# Scenario: Small lookup table needed on all executors
lookup = {"US": "United States", "UK": "United Kingdom"}

# Without broadcast: Sends to each partition (wasteful)
# With broadcast: Send once to each executor

broadcast_lookup = spark.broadcast(lookup)

def enrich_country(code):
    return broadcast_lookup.value.get(code, code)

df.withColumn("country_name", enrich_country(df['country_code']))

# Extract
df = spark.read.parquet("s3://bucket/data/*.parquet")

# Transform
df_clean = df.dropna()
df_clean = df_clean.filter(df_clean['amount'] > 0)
df_clean = df_clean.withColumn('amount_usd', df['amount'] * 1.1)

# Load
df_clean.write.mode("overwrite").parquet("s3://bucket/output/")

from pyspark.sql.functions import col, window

# Read from Kafka
df_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "events") \
    .load()

# Parse JSON
events = df_stream.select(col("value").cast("string"))

# 5-minute sliding window
windowed = events.groupBy(window(col("timestamp"), "5 minutes")).count()

# Write to console (for testing)
query = windowed.writeStream \
    .format("console") \
    .option("checkpointLocation", "/tmp/checkpoint") \
    .start()

query.awaitTermination()

# Broadcast join (small table + large table)
users = spark.read.csv("users.csv")  # 1GB
events = spark.read.csv("events.csv")  # 100GB

# Spark automatically broadcasts small table
result = events.join(users, "user_id")  # Efficient

# Explicit broadcast for control
result = events.join(broadcast(users), "user_id")

# Shuffle join (both large) - slow
users2 = spark.read.csv("users2.csv")  # 50GB
result = events.join(users2, "user_id")  # Shuffle happens

Spark Streaming (micro-batches):
  Events arrive → Batch 1 (0-1s) → Process → Results
               → Batch 2 (1-2s) → Process → Results
  Minimum latency: 500ms (batch duration)

Flink (true streaming):
  Event 1 → Process immediately → Results (< 100ms)
  Event 2 → Process immediately → Results (< 100ms)
  Minimum latency: 1-10ms (event at a time)

Source Operators (Kafka, Files, Sockets)
        ↓
DataStream API (transformations)
        ↓
Sink Operators (HDFS, Kafka, DB)

from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession

# For Flink, use different imports
env = StreamExecutionEnvironment.get_execution_environment()

# Source
data_stream = env.add_source(KafkaSource(...))

# Transformations
processed = data_stream.map(lambda x: x * 2) \
                       .filter(lambda x: x > 100)

# Sink
processed.add_sink(KafkaSink(...))

# Execute
env.execute("Job Name")

from flink.datastream.window import TumblingEventTimeWindows
from flink.datastream.functions import AggregateFunction
import time

# Window types
# Tumbling: Fixed size, no overlap
# [0-60s] [60-120s] [120-180s]
windowed = stream.window(TumblingEventTimeWindows.of(Time.seconds(60)))

# Sliding: Overlap
# [0-60s] [30-90s] [60-120s]
windowed = stream.window(SlidingEventTimeWindows.of(
    Time.seconds(120),  # window size
    Time.seconds(30)    # slide by
))

# Session: Triggered by inactivity
# [events] <gap> [events] <gap> [events]
windowed = stream.window(SessionWindow(Time.minutes(5)))

from flink.datastream.state import MapState, AggregateFunction

class CountingFunction(AggregateFunction):
    def create_accumulator(self):
        return 0  # Start state
    
    def add(self, value, accumulator):
        return accumulator + value  # Update state
    
    def get_result(self, accumulator):
        return accumulator  # Return state
    
    def merge(self, a, b):
        return a + b  # Merge two states

# Use in window
stream.keyBy(lambda x: x['user_id']) \
      .window(TumblingEventTimeWindows.of(Time.minutes(5))) \
      .aggregate(CountingFunction())

Without exactly-once:
  Kafka → Flink (process) → DB
  System crashes after processing, before DB write
  Result: Event lost

  Flink → DB → Crash
  Event already in DB, but retry sends again
  Result: Duplicate

With exactly-once:
  Flink checkpoints state
  On restart, resume from checkpoint
  No duplicates, no loss

env = StreamExecutionEnvironment.get_execution_environment()

# Enable checkpointing (exactly-once)
env.enable_checkpointing(60000)  # Checkpoint every 60s
env.get_checkpoint_config() \
    .set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)

# Idempotent sink required for exactly-once
stream.add_sink(IdempotentKafkaSink(...))

class FraudDetectionFunction(KeyedProcessFunction):
    def open(self, runtime_context):
        # State: last seen timestamp, count
        self.timer_state = runtime_context.get_state(...)
    
    def process_element(self, value, ctx):
        user_id = value['user_id']
        amount = value['amount']
        
        # Check rules
        if amount > 10000:
            yield ('fraud', value)
        elif amount > 5000:
            # Ask for 2FA
            yield ('verify', value)
        else:
            yield ('allow', value)

# Apply
stream.keyBy(lambda x: x['user_id']) \
      .process(FraudDetectionFunction())

# Example: User session analytics
# Session ends after 5 minutes of inactivity

stream.keyBy(lambda x: x['user_id']) \
      .window(SessionWindow(Time.minutes(5))) \
      .apply(lambda session_events: {
          'user_id': session_events[0]['user_id'],
          'session_duration': len(session_events),
          'total_actions': len(session_events),
          'total_revenue': sum(e['amount'] for e in session_events)
      })

# Stream: User clicks
# Batch: Product catalog (side input)

from flink.datastream.functions import CoFlatMapFunction

class EnrichedClicksFunction(CoFlatMapFunction):
    def flat_map1(self, value):
        # Stream processing (clicks)
        yield value
    
    def flat_map2(self, value):
        # Side input (catalog updates)
        # Cache/update product info
        cache[value['product_id']] = value

# Implement enrichment
clicks.connect(catalog) \
      .flat_map(EnrichedClicksFunction())

# Start cluster
./bin/start-cluster.sh

# Run job
./bin/flink run -d -c MyJob MyJob.jar

# Monitor
http://localhost:8081 (web UI)

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: fraud-detection
spec:
  image: flink:latest
  flinkVersion: v1_18
  replicas: 3
  taskManager:
    replicas: 5
  jarSpec:
    jarURI: s3://bucket/fraud-detection.jar
  entrypoint: com.example.FraudDetectionJob

Need real-time analytics? 
├─ YES, < 100ms latency → Flink ✓
├─ YES, 500ms-1s OK → Spark Streaming ✓
└─ NO, batch jobs → Spark ✓

Exactly-once semantics critical?
├─ YES → Flink ✓
└─ NO → Spark Streaming ✓

Complex state management?
├─ YES → Flink ✓
└─ NO → Spark ✓

Need unified batch+stream?
├─ YES → Spark (but not true streaming) ✓
└─ YES (pure stream) → Flink ✓

Team expertise?
├─ Spark → Spark ✓
├─ Kafka → Kafka Streams or Flink ✓
└─ New → Start with Spark, migrate if needed

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'data_team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'etl_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # 2 AM daily
    start_date=datetime(2024, 1, 1),
)

def extract_data():
    # Query API
    import requests
    data = requests.get('https://api.example.com/events').json()
    return data

def transform_data(ti):
    # Get data from upstream task
    data = ti.xcom_pull(task_ids='extract')
    # Transform: filter, aggregate, clean
    transformed = [d for d in data if d['valid']]
    ti.xcom_push(key='cleaned_data', value=transformed)

def load_data(ti):
    data = ti.xcom_pull(task_ids='transform', key='cleaned_data')
    # Load to data warehouse
    import psycopg2
    conn = psycopg2.connect("dbname=warehouse")
    cur = conn.cursor()
    for row in data:
        cur.execute("INSERT INTO events VALUES (%s, %s)", row)
    conn.commit()

extract = PythonOperator(
    task_id='extract',
    python_callable=extract_data,
    dag=dag,
)

transform = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    dag=dag,
)

load = PythonOperator(
    task_id='load',
    python_callable=load_data,
    dag=dag,
)

dq_check = BashOperator(
    task_id='data_quality_check',
    bash_command='python /scripts/validate.py',
    dag=dag,
)

# Define dependencies
extract >> transform >> load >> dq_check

from prefect import flow, task
from prefect.tasks.bash import bash_shell
import requests

@task(retries=2, retry_delay_seconds=300)
def extract_data():
    response = requests.get('https://api.example.com/events')
    response.raise_for_status()
    return response.json()

@task
def transform_data(data):
    # Type hints help Prefect understand data flow
    return [d for d in data if d['valid']]

@task
def load_data(data):
    # Prefect handles logging automatically
    print(f"Loading {len(data)} records")
    # Load to warehouse
    pass

@flow
def etl_pipeline():
    raw_data = extract_data()
    clean_data = transform_data(raw_data)
    load_data(clean_data)

# Schedule: every day at 2 AM
if __name__ == "__main__":
    etl_pipeline()

from dagster import job, op, In, Out

@op
def extract_data():
    import requests
    return requests.get('https://api.example.com/events').json()

@op
def transform_data(data):
    return [d for d in data if d['valid']]

@op
def load_data(data):
    print(f"Loaded {len(data)} records to warehouse")

@job
def etl_job():
    load_data(transform_data(extract_data()))

if __name__ == "__main__":
    etl_job.execute_in_process()

{
  "Comment": "ETL pipeline",
  "StartAt": "ExtractData",
  "States": {
    "ExtractData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:extract",
      "Next": "TransformData",
      "Retry": [{
        "ErrorEquals": ["States.TaskFailed"],
        "IntervalSeconds": 300,
        "MaxAttempts": 2,
        "BackoffRate": 1.0
      }]
    },
    "TransformData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:transform",
      "Next": "LoadData"
    },
    "LoadData": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:putItem",
      "Parameters": {
        "TableName": "processed_events",
        "Item": {
          "id": {"S.$": "$.id"},
          "data": {"S.$": "$.data"}
        }
      },
      "End": true
    }
  }
}

import luigi
import requests
import pandas as pd

class ExtractData(luigi.Task):
    def output(self):
        return luigi.LocalTarget('data/raw.json')
    
    def run(self):
        data = requests.get('https://api.example.com/events').json()
        with self.output().open('w') as f:
            import json
            json.dump(data, f)

class TransformData(luigi.Task):
    def requires(self):
        return ExtractData()
    
    def output(self):
        return luigi.LocalTarget('data/transformed.csv')
    
    def run(self):
        df = pd.read_json(self.input().path)
        df = df[df['valid'] == True]
        df.to_csv(self.output().path, index=False)

class LoadData(luigi.Task):
    def requires(self):
        return TransformData()
    
    def run(self):
        df = pd.read_csv(self.input().path)
        # Load to warehouse
        print(f"Loaded {len(df)} rows")

if __name__ == '__main__':
    luigi.build([LoadData()], local_scheduler=True)

START
├─ Is it AWS-only? → YES → Use Step Functions ✓
├─ Need to deploy today?
│  ├─ YES → Use Luigi or cron ✓
│  ├─ NO → Continue
├─ 100+ task dependencies?
│  ├─ YES → Use Airflow ✓
│  ├─ NO → Continue
├─ Prefer modern UI/UX?
│  ├─ YES → Use Prefect ✓
│  ├─ NO → Continue
├─ Need data asset tracking?
│  ├─ YES → Use Dagster ✓
│  ├─ NO → Airflow ✓

Setup: 2 weeks (MetaDB, Scheduler, config)
Code: 150 lines Python
Monitoring: Excellent UI
Operations: Requires ops team
Cost: Infrastructure + personnel
Scalability: Handles 1000+ tasks

Setup: 2 days (sign up, deploy)
Code: 100 lines Python (simpler)
Monitoring: Modern UI (cloud)
Operations: Managed (Prefect handles)
Cost: Lower (no infrastructure)
Scalability: 500+ tasks

Setup: 1 day
Code: 200 lines JSON
Monitoring: AWS CloudWatch
Operations: Fully managed
Cost: $0.25 per 1M executions
Scalability: 1000+ concurrent

Setup: Hours
Code: 50 lines Python
Monitoring: CLI only
Operations: Manual
Cost: Minimal
Scalability: 100 tasks max

Raw data → Speed layer (stream)  → Real-time results
        → Batch layer (batch)    → Batch results
        ↓
    Serving layer (merge results)

Raw data → Stream processing → Results

Late data → Re-stream from data lake → Re-process

Transaction → Kafka topic
           ↓
    Stream processor
    (checks against rules)
    - Amount > $10,000?
    - Location change > 1000km in 1 hour?
    - Unusual merchant?
           ↓
    Score: 0.0 (safe) to 1.0 (fraud)
           ↓
    If score > 0.9 → Block transaction
    If 0.5-0.9 → Ask for 2FA
    If < 0.5 → Allow

Batch (hourly):
  9:00 AM: Transaction happens (fraud)
  10:00 AM: Batch job detects fraud
  1 HOUR LATE! Fraud already processed

Stream (real-time):
  9:00 AM: Transaction happens
  9:00:100ms: Detected as fraud
  Blocked instantly

env = StreamExecutionEnvironment.get_execution_environment()

transactions = env.add_source(KafkaSource(...))

fraud_scores = transactions.map(calculate_fraud_score)
               .filter(lambda x: x['score'] > 0.5)
               .sink_to(BlockedTransactionsSink())

env.execute("Fraud Detection")

Day 1: 9 PM
  All events for Day 1 accumulated in data lake
  
Day 1: 11 PM
  Batch job starts:
  - Read 1TB of events
  - Group by user, device, country
  - Generate metrics
  - Load into data warehouse
  
Day 2: 12 AM
  Reports available (1 hour after day ends)
  Analytics team reviews

Stream (real-time):
- Process 1M events/sec continuously
- High CPU/memory cost: 24/7
- Overkill for next-day reports

Batch (12 hours at night):
- Process 86.4B events in 12-hour window
- Cost: 1/2 (run half the day)
- Simpler to debug (re-run if error)

Stream: 1000 instances × $0.10/hour × 24 hours = $2,400/day
Batch: 1000 instances × $0.10/hour × 12 hours = $1,200/day

Savings: $1,200/day or $400K/year
Plus: Simpler to maintain, easier to correct errors

Real-time layer (stream):
  - User views product X
  - Immediately recommend similar
  - Based on live session

Batch layer:
  - Daily: Recalculate models
  - Feedback: Which recommendations worked?
  - Update ML models

Serving layer:
  - Merge both sources
  - Pick best recommendations

User browsing electronics → Stream processor
    "User browsing laptops"
        ↓
    Real-time: Show similar laptops (instant)
        ↓
    Batch (daily): Did user buy? Feedback to model
        ↓
    Next day: Model improves (recommendation better)

# Stream layer
def real_time_recommend(user_id, product_id):
    similar = redis.get(f"similar:{product_id}")
    return similar or default_similar  # Fast

# Batch layer
def train_recommendations_daily():
    # Read all transactions from yesterday
    feedback = get_yesterday_transactions()
    
    # Train ML model
    model = train_ml_model(feedback)
    
    # Store precomputed recommendations
    for product in all_products:
        similar = model.predict_similar(product)
        redis.set(f"similar:{product}", similar)

# Serving layer
def get_recommendations(user_id, product_id):
    return real_time_recommend(user_id, product_id)
    # Or: merge with batch ML scores if needed

Old: Process all 1TB at 9 PM (6 hours)
New: 
  - 8 PM: Process only CHANGES since last run (100GB)
  - 2 PM: Process CHANGES since 8 PM (50GB)
  - 6 PM: Process CHANGES since 2 PM (30GB)
  - 9 PM: Process remaining changes (20GB)

Result: Reports every 2 hours (not 6)

New: Process 50% sample of data (3 hours)
Result: ~95% accuracy in 3 hours

Better than 6 hours of full accuracy

Move to stream processing
Real-time aggregation
Reports always current

Complexity: Need to rewrite pipeline

def batch_job():
    last_checkpoint = redis.get("batch:checkpoint") or "2024-01-01"
    
    # Process only new data
    new_events = db.query(f"""
        SELECT * FROM events 
        WHERE timestamp > {last_checkpoint}
    """)
    
    # Aggregate
    metrics = process(new_events)
    
    # Store results
    results_db.insert(metrics)
    
    # Save checkpoint
    redis.set("batch:checkpoint", now())
    
    # Next run: start from here (50GB instead of 1TB)

Events → Kafka (buffering)
      ↓
Stream processor (Apache Flink)
  - Aggregate by (user, device, country)
  - Window: 1 minute
  - Calculate metrics
      ↓
Columnar store (Druid or ClickHouse)
  - Optimized for analytics queries
  - High cardinality (billions of values)
  - Fast range scans
      ↓
User queries
  "Show traffic by country last hour"
  "Revenue by device type today"

Row-oriented (traditional DB):
Row 1: user_id, device, country, revenue, timestamp
Row 2: user_id, device, country, revenue, timestamp
Query: "Sum revenue by country"
→ Must read entire rows (includes user_id, device, timestamp you don't need)

Columnar (Druid):
Column: [user_id, user_id, user_id, ...]
Column: [device, device, device, ...]
Column: [country, country, country, ...]
Column: [revenue, revenue, revenue, ...]
Query: "Sum revenue by country"
→ Only read country and revenue columns (ignore user_id, device, timestamp)
→ 10x faster

1M events/sec → Kafka → Flink stream job (aggregation)
                              ↓
                        Micro-batches (1 sec windows)
                              ↓
                        Druid (columnar index)
                              ↓
                        Analytics queries (drill down by any dimension)

Traditional: 1M events/sec → PostgreSQL
Cost: Expensive (row-oriented not suited for this)

Columnar: 1M events/sec → Druid
Cost: 1/10th (columnar compression + efficient scans)

Result: Real-time analytics at scale, affordable