# Lab: Introduction to Spark's RDD API

## Summary

- Built RDDs from the bronze datasets (`users.csv`, `orders.csv`) and looked at their partitions.
- Observed lazy evaluation and lineage with `toDebugString()` and the Spark UI.
- Handled the multiline address problem of `textFile()` by filtering lines that start with a UUID.
- Parsed orders by hand, aggregated with `reduceByKey` and `groupByKey`, deduplicated orders and joined them with users.

## Questions

### RDD creation and partitions

**What decides the number of partitions `textFile()` picks by default?**
`textFile()` asks for a minimum number of partitions (at most 2 by default), and Hadoop splits the file by block size
(128 MB). A small file like this one gives 2 partitions.

**Which of `repartition()` and `coalesce()` shuffles data?**
`repartition()` does a full shuffle. `coalesce()` merges existing partitions without a shuffle, so it is cheaper, but
it can only reduce the number of partitions by default.

### Lazy evaluation and lineage

**Does a second `orders_data.count()` create a new stage in the Spark UI? Why?**
Yes. The RDD is not cached, so Spark recomputes it from the file at each action.

**When would `.cache()` change that answer?**
If `.cache()` is called on `orders_data` before the first `count()`, the first action fills the cache and the second one
reads from memory, with no new read stage.

### The multiline address

**Why does `spark.read.csv(multiLine=True)` not have this problem, while `sc.textFile()` does?**
The CSV parser understands quotes, so a newline inside a quoted field does not end the record. `textFile()` splits on
every newline, whatever the quotes.

**What must be true of a file format for a line-oriented reader to be safe?**
Each record must be exactly one physical line, so no field can contain a line break.

### Manual parsing of orders

**What happens if one line has an extra comma inside `product`?**
`split(",")` returns 6 values instead of 5, and the unpacking raises a `ValueError`. Because RDDs are lazy, the job
only fails at the action that processes that line.

**Where does the DataFrame API perform the equivalent check, and when?**
The CSV reader checks rows against a schema while reading, with the `mode` option (`PERMISSIVE`, `DROPMALFORMED`,
`FAILFAST`). It also runs at the action, not at definition time.

### reduceByKey vs groupByKey

**What does `groupByKey` shuffle, and why is it more expensive?**
It shuffles every `(key, value)` pair with no local combining. `reduceByKey` first reduces each partition, so only
one partial value per key and per partition crosses the network. The Spark UI shows a smaller shuffle write and read
for `reduceByKey`.

**When is `groupByKey` unavoidable?**
When all the values of a key are needed together, for example to compute a median or a percentile, or to sort the
values inside a group.

### Manual deduplication

**What plays the role of `partitionBy()` here?**
The shuffle done by `reduceByKey` on the `order_id` key. It brings every row of one `order_id` to the same partition.

**Which version states the intent more directly?**
For "keep the most recent", the `reduceByKey` with a custom combiner is short and direct, but the combiner must be
associative and commutative. The window function is more declarative and generalizes better, for example to the top N
rows per group.

### Joining orders with users

**What must be true for Spark to skip the shuffle?**
Both RDDs must already use the same partitioner with the same number of partitions (for example `partitionBy(n)` on
both, then persisted).

**What changes with `leftOuterJoin()` if a `user_id` has no matching user?**
The order is kept as `(user_id, (order, None))` instead of being dropped.

**What is the risk of `.collect()` on `enriched` with a much larger dataset?**
All the data is brought into the driver's memory, which can crash it. Use `take()`, `count()` or write the result to
files instead.

## Exercises

**1. `aggregateByKey`**

```python
stats = (
    orders_rdd.map(lambda o: (o["user_id"], o["quantity"]))
    .aggregateByKey(
        (0, 0),
        lambda acc, q: (acc[0] + 1, acc[1] + q),
        lambda a, b: (a[0] + b[0], a[1] + b[1]),
    )
)
avg_quantity = stats.mapValues(lambda s: s[1] / s[0])
```

**2. Orders with no matching user**

```python
fake = sc.parallelize([("fake-user", {"order_id": "x", "user_id": "fake-user",
                                      "date": "2020-01-01", "quantity": 1, "product": "bread"})])
orders_fake_kv = orders_rdd.map(lambda o: (o["user_id"], o)).union(fake)
left = orders_fake_kv.leftOuterJoin(users_kv)
print(left.filter(lambda kv: kv[1][1] is None).count())  # 1
```

**3. DataFrame version**

```python
orders_df = spark.read.csv("orders.csv", header=True, inferSchema=True)
orders_df.groupBy("user_uuid").sum("quantity").explain()
```

One line instead of several, and Spark optimizes the plan (partial then final aggregation), whereas the RDD lineage
shows exactly what was written by hand.

**4. Cache and timing**

With `.cache()`, the second run of `count()` is faster and the Storage tab of the Spark UI lists the RDD. Without it,
every action recomputes the parsing and the Storage tab stays empty.