MapReduce (2004-2012) enabled:
✅ Processing petabytes of data
✅ Fault tolerance on commodity hardware
✅ Simple programming model (just write map/reduce)
✅ Automatic parallelization
✅ Data locality optimization

Used successfully at: Google, Yahoo, Facebook, LinkedIn

MapReduce WordCount:
┌─────────┐
│  Input  │ (Read from HDFS)
│  Data   │
└────┬────┘
     │
     ▼
┌─────────┐
│   Map   │ (Process in memory)
└────┬────┘
     │
     ▼ WRITE TO DISK (shuffle)
┌─────────┐
│Local Disk│
└────┬────┘
     │
     ▼ READ FROM DISK
┌─────────┐
│ Shuffle │ (Network + disk)
└────┬────┘
     │
     ▼ WRITE TO DISK
┌─────────┐
│ Reduce  │
└────┬────┘
     │
     ▼ WRITE TO DISK (output)
┌─────────┐
│  HDFS   │
└─────────┘

Total disk writes: 3x data size
Total disk reads: 2x data size

// Machine Learning: Logistic Regression
// Needs 10-20 iterations over same dataset

MapReduce approach (pseudo-code):
for (i <- 1 to 20) {
  // Read from HDFS
  data = loadFromHDFS("training_data")

  // Run map-reduce iteration
  gradient = data.map(computeGradient).reduce(sum)
  updateWeights(gradient)

  // Write back to HDFS
  saveToHDFS(weights)
}

// Problem: Reads entire dataset from disk 20 times!
// 100GB dataset × 20 iterations = 2TB disk reads
// On a cluster: Hours of wasted time

Data scientist workflow:
1. Load 1TB dataset from HDFS (2 minutes)
2. Run query 1 → Wait 5 minutes
3. Run query 2 on same data → Wait 5 minutes again!
4. Run query 3 → Wait 5 minutes again!

Why? MapReduce can't keep data in memory between queries

Spark approach to WordCount:
┌─────────┐
│  Input  │ (Read from HDFS once)
│  Data   │
└────┬────┘
     │
     ▼ STAYS IN MEMORY
┌─────────┐
│   RDD   │ (Distributed across cluster RAM)
│  (Map)  │
└────┬────┘
     │
     ▼ STAYS IN MEMORY (no disk!)
┌─────────┐
│   RDD   │
│ (Reduce)│
└────┬────┘
     │
     ▼ WRITE TO DISK (only final result)
┌─────────┐
│  HDFS   │
└─────────┘

Total disk writes: 1x data size (final output only)
Total disk reads: 1x data size (initial input only)

Replicate data in memory across nodes
Problem: Wastes 2/3 of memory

// Example: Processing log files
val logs = sc.textFile("hdfs://logs/*.txt")  // RDD 1
val errors = logs.filter(_.contains("ERROR")) // RDD 2
val counts = errors.map(extractCode).reduceByKey(_ + _) // RDD 3

// Lineage graph:
RDD 3 (counts) ← depends on RDD 2 (errors)
RDD 2 (errors) ← depends on RDD 1 (logs)
RDD 1 (logs)   ← depends on HDFS

// If RDD 2's partition on node5 is lost:
// Just recompute: logs.filter(_.contains("ERROR"))
// No need to store RDD 2 redundantly!

Original Computation:
Node 1: [logs_part_1] → filter → [errors_part_1] → map/reduce → result_1
Node 2: [logs_part_2] → filter → [errors_part_2] → map/reduce → result_2
Node 3: [logs_part_3] → filter → [errors_part_3] → map/reduce → result_3

Node 2 crashes! (loses errors_part_2)

Recovery:
Node 4: Re-reads logs_part_2 from HDFS
        → Re-applies filter
        → Recreates errors_part_2
        → Continues computation

Total overhead: Only recompute lost partition, not entire RDD

// Narrow transformations (no shuffle needed)
map(f: T => U)           // Apply f to each element
filter(f: T => Boolean)  // Keep elements where f returns true
flatMap(f: T => Seq[U])  // Map then flatten results
mapPartitions(f)         // Apply f to entire partition at once

// Wide transformations (require shuffle)
groupByKey()             // Group values by key (avoid if possible!)
reduceByKey(f)           // Reduce values per key (preferred over groupByKey)
sortByKey()              // Sort RDD by key
join(other)              // Join two RDDs by key
cogroup(other)           // Group multiple RDDs by key

val data = sc.textFile("huge_file.txt")     // ① Not executed yet!
val filtered = data.filter(_.contains("ERROR")) // ② Still not executed!
val mapped = filtered.map(_.length)         // ③ Still not executed!

// Only when an action is called:
val result = mapped.collect()  // ④ NOW everything executes!

// Spark builds execution plan:
// textFile → filter → map → collect
// Then optimizes and executes all at once

// User writes:
data.filter(x => x > 10).map(x => x * 2)

// Spark can optimize to:
data.mapPartitions { partition =>
  partition.filter(x => x > 10).map(x => x * 2)
}
// Fuses operations, avoids intermediate RDD materialization

val data = sc.textFile("1TB_file.txt")
val processed = data.map(expensiveOperation)  // Not executed
val sample = processed.take(10)  // Only processes enough to get 10 items!

// Return data to driver
reduce(f: (T, T) => T)   // Aggregate all elements
collect()                 // Return all elements to driver (dangerous for big data!)
count()                   // Count number of elements
first()                   // Return first element
take(n)                   // Return first n elements
takeSample(n)             // Random sample of n elements

// Write to storage
saveAsTextFile(path)      // Write as text files
saveAsSequenceFile(path)  // Write as Hadoop SequenceFile
saveAsObjectFile(path)    // Serialize objects to file

// Side effects
foreach(f: T => Unit)     // Apply f to each element (for side effects)
foreachPartition(f)       // Apply f to each partition

// DANGER: Don't collect large RDDs!
val bigData = sc.textFile("1PB_of_data.txt")
val result = bigData.collect()  // ❌ Will crash driver with OutOfMemoryError!

// Instead: Use distributed operations
val count = bigData.count()     // ✅ Computed on cluster, only count returned
val sample = bigData.take(100)  // ✅ Only 100 items to driver

import org.apache.spark.storage.StorageLevel

// Common persistence levels
MEMORY_ONLY          // Store deserialized in JVM heap (fastest, most memory)
MEMORY_AND_DISK      // Spill to disk if memory full
MEMORY_ONLY_SER      // Store serialized (saves space, slower)
DISK_ONLY            // Store only on disk
OFF_HEAP             // Store in off-heap memory (Tachyon/Alluxio)
MEMORY_AND_DISK_SER  // Serialized in memory, spill to disk

// Usage example
val important = data.filter(_.important)
important.persist(StorageLevel.MEMORY_AND_DISK)
// Now 'important' will be kept in memory for reuse

// Or use cache() shorthand for MEMORY_ONLY
important.cache()

// Don't forget to unpersist when done!
important.unpersist()

Is data reused multiple times?
├─ No → Don't persist
└─ Yes ↓

   Does it fit in memory?
   ├─ Yes → MEMORY_ONLY (fastest)
   └─ No ↓

      Can you afford to recompute if evicted?
      ├─ Yes → MEMORY_ONLY (let it recompute)
      └─ No → MEMORY_AND_DISK

         Is serialization overhead acceptable?
         ├─ No → Keep deserialized
         └─ Yes → Use _SER variants (save memory)

abstract class RDD[T] {
  // Get partitions
  def partitions: Array[Partition]

  // Compute a partition given parent partitions
  def compute(partition: Partition, context: TaskContext): Iterator[T]

  // List of parent RDDs
  def dependencies: Seq[Dependency[_]]

  // Partitioner (optional, for key-value RDDs)
  def partitioner: Option[Partitioner]

  // Preferred locations for a partition (data locality)
  def preferredLocations(partition: Partition): Seq[String]
}

class FilteredRDD[T](parent: RDD[T], f: T => Boolean) extends RDD[T] {

  override def partitions = parent.partitions

  override def compute(partition: Partition, context: TaskContext) = {
    parent.iterator(partition, context).filter(f)
  }

  override def dependencies = List(new OneToOneDependency(parent))
}

// Examples
map, filter, union, join with co-partitioned inputs

// Visualization
Parent RDD:  [P1] [P2] [P3] [P4]
                ↓    ↓    ↓    ↓
Child RDD:   [C1] [C2] [C3] [C4]

// Each child partition depends on ≤ 1 parent partition

// Examples
groupByKey, reduceByKey, join with non-co-partitioned inputs

// Visualization
Parent RDD:  [P1] [P2] [P3] [P4]
               ↓↘  ↓↗↘  ↓↗  ↓
Child RDD:    [C1]    [C2]   [C3]

// Each child partition depends on MULTIPLE parent partitions

// Narrow: Only recompute lost partition
Parent: [P1] [P2] [P3]
           ↓    ✗    ↓
Child:  [C1] [C2] [C3]

// If C2 lost: Only recompute P2 → C2

// Wide: Must recompute from multiple parents
Parent: [P1] [P2] [P3]
          ↓↘  ↓↗  ↓
Child:     [C1] [✗]

// If C2 lost: Need data from P1, P2, P3
// If parents not cached: Recompute all!

// Narrow transformations can be pipelined:
data.map(f1).filter(f2).map(f3)
// Executes as: x => f3(f2(f1(x)))
// One pass through data!

// Wide transformation creates stage boundary:
data.map(f1).groupByKey().map(f2)
//           ↑ STAGE 1 ↑ shuffle ↑ STAGE 2 ↑
// Two stages, shuffle writes/reads disk

// BAD: Wide dependency early
data.groupByKey().filter(hasMany)
// Shuffles ALL data, then filters

// GOOD: Narrow dependency early
data.filter(relevant).groupByKey()
// Filters first (no shuffle), then shuffles less data

// EVEN BETTER: Use reduceByKey instead of groupByKey
data.map(x => (x.key, 1)).reduceByKey(_ + _)
// Combines locally before shuffle (combiner pattern)

case class Lineage(
  rdd: RDD[_],
  dependency: Dependency[_],
  parent: Option[Lineage]
)

// Example lineage:
val logs = sc.textFile("input.txt")
val errors = logs.filter(_.contains("ERROR"))
val counts = errors.map(x => (x, 1)).reduceByKey(_ + _)

// Lineage graph:
counts.toDebugString
/*
(2) ShuffledRDD[2] at reduceByKey
 +-(2) MapPartitionsRDD[1] at map
    +-(2) MapPartitionsRDD[0] at filter
       +-(2) input.txt MapPartitionsRDD
*/

Algorithm: RecoverPartition(rdd, partition)

1. If partition is cached and available:
     Return cached data

2. Else if partition's parent RDDs are available:
     parentData = for each parent in rdd.dependencies:
                    RecoverPartition(parent.rdd, parent.partition)
     Return rdd.compute(partition, parentData)

3. Else:
     Recursively recover parent partitions first
     Then compute this partition

Optimization: Only recompute lost partitions, not entire RDD

Scenario: 1000-partition RDD, 1 partition lost

Replication approach:
- Storage: 1000 partitions × 3 replicas = 3000 partition-copies
- Recovery: Read from replica (instant)

Lineage approach:
- Storage: Only lineage graph (< 1KB typically)
- Recovery: Recompute 1 partition from source

Winner: Lineage (for most workloads)
Exception: Very long lineage chains → use checkpointing

Hadoop MapReduce:
- Time: 127 minutes
- Why slow: Reads 100GB from HDFS × 100 iterations = 10TB disk reads

Spark (data cached in memory):
- First iteration: 80 seconds (read from HDFS)
- Subsequent iterations: 1 second each
- Total time: 80s + 99s = 179 seconds ≈ 3 minutes
- Speedup: 42x

Spark (without caching - recomputes each time):
- Time: ~80 minutes
- Speedup: 1.6x (still faster due to less disk I/O overhead)

Hadoop:
- Time: 171 seconds per iteration
- Total: 1710 seconds
- Bottleneck: Shuffle and disk I/O

Spark:
- Time: 23 seconds per iteration (with caching)
- Total: 230 seconds
- Speedup: 7.4x

Why smaller speedup than logistic regression?
- PageRank has wide dependencies (join)
- Shuffle still requires disk in Spark
- But in-memory between iterations helps

// PageRank in Spark (simplified)
var ranks = links.mapValues(v => 1.0)
links.cache() // Cache link structure

for (i <- 1 to 10) {
  val contribs = links.join(ranks).flatMap {
    case (url, (links, rank)) =>
      links.map(dest => (dest, rank / links.size))
  }
  ranks = contribs.reduceByKey(_ + _).mapValues(0.15 + 0.85 * _)
}

// Key: links RDD is cached and reused 10 times
// Only ranks RDD is recomputed each iteration

// Load and cache data
val wiki = sc.textFile("hdfs://wikipedia/*.txt").cache()

// Query 1: Pages mentioning "machine learning"
wiki.filter(_.contains("machine learning")).count()
// First time: 170s (read from HDFS)

// Query 2: Pages with "deep learning"
wiki.filter(_.contains("deep learning")).count()
// Cached: 5s

// Query 3: Average page length
wiki.map(_.length).mean()
// Cached: 7s

// Query 4: Most common words
wiki.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).top(100)
// Cached: 15s

Hadoop (no caching):
- Query 1: 170s
- Query 2: 170s (re-reads from HDFS!)
- Query 3: 170s (re-reads again!)
- Total: 510s+

Spark (with caching):
- Query 1: 170s (initial load)
- Query 2: 5s (from cache)
- Query 3: 7s (from cache)
- Total: 182s
- Speedup: 2.8x (and improves with more queries)

Spark vs Hadoop:
- Hadoop: 30 minutes
- Spark: 2.8 minutes
- Speedup: 10.7x

Why effective:
- Dataset cached in memory
- 30 iterations × 100GB = 3TB saved disk reads

┌─────────────────────────────────────────────────────────────┐
│                    SPARK APPLICATION                        │
│                                                             │
│  ┌──────────────────────────────────────────────────┐      │
│  │         Driver Program (SparkContext)            │      │
│  │                                                  │      │
│  │  ┌──────────────┐  ┌─────────────┐             │      │
│  │  │ DAGScheduler │  │TaskScheduler│             │      │
│  │  │   - Builds   │  │  - Assigns  │             │      │
│  │  │     stages   │  │    tasks to │             │      │
│  │  │   - Tracks   │  │   executors │             │      │
│  │  │   lineage    │  └─────────────┘             │      │
│  │  └──────────────┘                               │      │
│  │                                                  │      │
│  │  ┌──────────────────────────────┐               │      │
│  │  │    Block Manager Master      │               │      │
│  │  │  - Tracks RDD block locations│               │      │
│  │  └──────────────────────────────┘               │      │
│  └──────────────────┬───────────────────────────────┘      │
│                     │                                       │
│          Submits tasks & tracks status                     │
│                     │                                       │
│                     ▼                                       │
│  ┌────────────────────────────────────────────────────┐    │
│  │         Cluster Manager                            │    │
│  │  (YARN / Mesos / Standalone / Kubernetes)          │    │
│  │  - Resource allocation                             │    │
│  │  - Executor lifecycle management                   │    │
│  └────────────────┬───────────────────────────────────┘    │
│                   │                                         │
│         ┌─────────┼─────────┬─────────┐                    │
│         │         │         │         │                    │
│         ▼         ▼         ▼         ▼                    │
│  ┌──────────┐┌──────────┐┌──────────┐┌──────────┐         │
│  │Executor 1││Executor 2││Executor 3││Executor N│         │
│  │          ││          ││          ││          │         │
│  │┌────────┐││┌────────┐││┌────────┐││┌────────┐│         │
│  ││Task    │││Task     │││Task     │││Task     ││         │
│  ││Runner  │││Runner   │││Runner   │││Runner   ││         │
│  │└────────┘││└────────┘││└────────┘││└────────┘│         │
│  │          ││          ││          ││          │         │
│  │┌────────┐││┌────────┐││┌────────┐││┌────────┐│         │
│  ││Block   │││Block    │││Block    │││Block    ││         │
│  ││Manager │││Manager  │││Manager  │││Manager  ││         │
│  │└────────┘││└────────┘││└────────┘││└────────┘│         │
│  │          ││          ││          ││          │         │
│  │  Cache:  ││  Cache:  ││  Cache:  ││  Cache:  │         │
│  │  RDD     ││  RDD     ││  RDD     ││  RDD     ││         │
│  │  blocks  ││  blocks  ││  blocks  ││  blocks  ││         │
│  └──────────┘└──────────┘└──────────┘└──────────┘         │
└─────────────────────────────────────────────────────────────┘

val data = sc.textFile("file.txt")          // RDD 1
  .map(_.toUpperCase)                        // RDD 2
  .filter(_.contains("ERROR"))               // RDD 3
  .map(line => (line.split(" ")(0), 1))      // RDD 4
  .reduceByKey(_ + _)                        // RDD 5
  .saveAsTextFile("output")                  // Action!

RDD DAG:
textFile → map → filter → map → reduceByKey → save
  (1)     (2)    (3)     (4)       (5)

Dependencies:
RDD 2: Narrow dependency on RDD 1
RDD 3: Narrow dependency on RDD 2
RDD 4: Narrow dependency on RDD 3
RDD 5: Wide dependency on RDD 4 (shuffle!)

Stage 1 (narrow transformations - can pipeline):
  textFile → map → filter → map

Stage 2 (after shuffle):
  reduceByKey → save

Stage boundaries: Wide dependencies (shuffles)

Assume input has 100 HDFS blocks:

Stage 1:
  - 100 tasks (one per input partition)
  - Each task: read block → map → filter → map → shuffle write

Stage 2:
  - Default: 200 tasks (spark.sql.shuffle.partitions)
  - Each task: shuffle read → reduce → write output

Scheduling preferences:
1. PROCESS_LOCAL: Task on same executor caching the data
2. NODE_LOCAL: Task on same node as data
3. RACK_LOCAL: Task on same rack
4. ANY: No locality preference

Example:
Task 1: Needs partition 0
- HDFS block 0 is on node5
- Spark tries to schedule on node5
- If node5 busy, tries rack-local
- Last resort: any available node

Driver tracks:
- Task completion status
- Partial results
- Failures and retries
- Shuffle data location

On failure:
- Resubmit failed task
- If executor lost: Recompute lost RDD partitions using lineage
- If too many failures: Abort job

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.storage.StorageLevel

object LogAnalytics {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Log Analytics")
    val sc = new SparkContext(conf)

    // Load logs from HDFS
    val logs = sc.textFile("hdfs://logs/2024/*/*/*.log")

    // Parse logs (assuming Apache Common Log Format)
    case class LogEntry(
      ip: String,
      timestamp: String,
      method: String,
      url: String,
      status: Int,
      bytes: Long
    )

    def parseLine(line: String): Option[LogEntry] = {
      val pattern = """^(\S+) \S+ \S+ \[([^\]]+)\] "(\S+) (\S+) \S+" (\d+) (\d+)""".r
      line match {
        case pattern(ip, ts, method, url, status, bytes) =>
          Some(LogEntry(ip, ts, method, url, status.toInt, bytes.toLong))
        case _ => None
      }
    }

    val parsed = logs.flatMap(parseLine)
    parsed.cache() // Cache for multiple analyses

    // Analysis 1: Error rate
    val totalRequests = parsed.count()
    val errors = parsed.filter(_.status >= 400).count()
    val errorRate = errors.toDouble / totalRequests
    println(s"Error rate: ${errorRate * 100}%")

    // Analysis 2: Top 10 URLs by traffic
    val topUrls = parsed
      .map(entry => (entry.url, entry.bytes))
      .reduceByKey(_ + _)
      .sortBy(_._2, ascending = false)
      .take(10)

    topUrls.foreach { case (url, bytes) =>
      println(s"$url: ${bytes / 1024 / 1024} MB")
    }

    // Analysis 3: Requests per hour
    val hourlyStats = parsed
      .map(entry => (entry.timestamp.substring(12, 14), 1))
      .reduceByKey(_ + _)
      .sortByKey()
      .collect()

    // Analysis 4: Client IP analysis
    val topIPs = parsed
      .map(entry => (entry.ip, 1))
      .reduceByKey(_ + _)
      .top(20)(Ordering.by(_._2))

    // Clean up
    parsed.unpersist()
    sc.stop()
  }
}

object PageRank {

  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setAppName("PageRank"))

    // Load links: each line is "source_url dest_url1 dest_url2 ..."
    val lines = sc.textFile("hdfs://pagerank/links.txt")

    // Parse into (source, Array[destinations])
    val links = lines.map { line =>
      val parts = line.split("\\s+")
      (parts(0), parts.tail)
    }.cache() // Cache because used in every iteration

    // Initialize ranks
    var ranks = links.mapValues(v => 1.0)

    // Iterate
    for (i <- 1 to 10) {
      // Calculate contributions from each page
      val contribs = links.join(ranks).flatMap {
        case (url, (links, rank)) =>
          val size = links.length
          links.map(dest => (dest, rank / size))
      }

      // Aggregate contributions and apply damping factor
      ranks = contribs.reduceByKey(_ + _).mapValues(0.15 + 0.85 * _)

      // Optional: Show progress
      if (i % 2 == 0) {
        println(s"Iteration $i complete")
      }
    }

    // Output top 10 pages
    val top = ranks.sortBy(_._2, ascending = false).take(10)
    top.foreach { case (url, rank) =>
      println(s"$url: $rank")
    }

    sc.stop()
  }
}

object PartitioningDemo {

  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setAppName("Partitioning"))

    // Create RDD with 4 partitions
    val data = sc.parallelize(1 to 1000, numSlices = 4)
    println(s"Number of partitions: ${data.getNumPartitions}")

    // Map partition index with its data
    val withPartitions = data.mapPartitionsWithIndex { (idx, iter) =>
      iter.map(value => (idx, value))
    }

    // Show partition distribution
    val partitionCounts = withPartitions
      .map { case (idx, _) => (idx, 1) }
      .reduceByKey(_ + _)
      .collect()
      .sorted

    partitionCounts.foreach { case (idx, count) =>
      println(s"Partition $idx: $count elements")
    }

    // Custom partitioning for key-value RDDs
    val keyValueRDD = data.map(x => (x % 10, x))

    // Default hash partitioning
    println(s"Default partitions: ${keyValueRDD.getNumPartitions}")

    // Custom partitioner: even/odd
    class EvenOddPartitioner(partitions: Int) extends Partitioner {
      def numPartitions: Int = partitions
      def getPartition(key: Any): Int = {
        val k = key.asInstanceOf[Int]
        if (k % 2 == 0) 0 else 1
      }
    }

    val customPartitioned = keyValueRDD.partitionBy(new EvenOddPartitioner(2))

    // Verify partitioning
    customPartitioned.mapPartitionsWithIndex { (idx, iter) =>
      Iterator((idx, iter.toList))
    }.collect().foreach { case (idx, elements) =>
      println(s"Partition $idx: ${elements.take(5)}")
    }

    sc.stop()
  }
}

2010: Spark research project begins at UC Berkeley
2012: RDD paper published at NSDI
2013: Spark becomes Apache incubator project
      Yahoo begins production deployment
2014: Spark becomes Apache top-level project
      Databricks founded by Spark creators
      Spark 1.0 released
2015: Spark surpasses Hadoop in popularity surveys
      Over 1000 contributors
2016: Spark 2.0 with structured APIs
      Databricks raises $140M
2018: Spark 3.0 development begins
2020: Spark 3.0 released (Adaptive Query Execution)
2024: Still the dominant big data framework
      10,000+ companies using in production

Systems Research: 40%
  - Distributed systems
  - Database systems
  - Operating systems

Machine Learning: 25%
  - Large-scale ML algorithms
  - Deep learning frameworks
  - AutoML systems

Data Management: 20%
  - Data lakes
  - ETL systems
  - Data quality

Cloud Computing: 10%
  - Serverless computing
  - Resource management
  - Autoscaling

Other: 5%
  - IoT, Edge computing, etc.

2012-2014: RDDs only
  - Low-level, flexible
  - Manual optimization needed
  - Type-safe but verbose

2015: DataFrames introduced
  - Higher-level API
  - Automatic optimization (Catalyst)
  - Better performance
  - But not type-safe in Scala

2016: Datasets (type-safe DataFrames)
  - Best of both worlds
  - Type-safe + optimized
  - Recommended for most use cases

2024 Recommendation:
  - Use DataFrames/Datasets by default
  - Use RDDs only for:
    * Unstructured data
    * Low-level control
    * Custom partitioning logic

var rdd = sc.textFile("input")
for (i <- 1 to 1000) {
  rdd = rdd.map(someTransformation)
}
// Lineage is 1000 transformations deep!
// If partition lost late: Expensive to recompute

Solution: Checkpoint periodically
rdd.checkpoint() // every 100 iterations

val grouped = data.groupByKey()  // Wide dependency
// If partition lost: Must re-shuffle!
// Shuffle data may be gone

Solution: Cache intermediate results
grouped.cache()

rdd.map(x => (x, Random.nextInt()))  // Non-deterministic!
// Recomputation gives different results
// Breaks lineage-based recovery

Solution: Use deterministic transformations

// cache() is shorthand for persist(MEMORY_ONLY)
rdd.cache()
// Same as:
rdd.persist(StorageLevel.MEMORY_ONLY)

// Use cache() when:
// - Data fits in memory
// - Recomputation is expensive
// - Data used multiple times

// Use persist(MEMORY_AND_DISK) when:
// - Data might not fit in memory
// - Don't want to recompute if evicted

// Use persist(MEMORY_ONLY_SER) when:
// - Data fits but memory tight
// - Can afford serialization overhead

// BEFORE (inefficient)
val data = sc.textFile("huge_file.txt")
val result = data
  .groupByKey()           // ❌ Wide shuffle
  .filter(_.size > 100)   // ❌ After shuffle
  .mapValues(_.sum)

// AFTER (optimized)
val result = data
  .filter(relevant)        // ✅ Filter early
  .mapValues(x => (x, 1))
  .reduceByKey { case ((sum1, cnt1), (sum2, cnt2)) =>
    (sum1 + sum2, cnt1 + cnt2)
  }                        // ✅ Use reduceByKey
  .filter { case (k, (sum, cnt)) => cnt > 100 }

// Improvements:
// 1. Filter before shuffle (less data to shuffle)
// 2. Use reduceByKey instead of groupByKey (local combine)
// 3. More efficient aggregation