Title: 544
Tags: lec24

Q: What is the difference between these Spark caching levels?
- MEMORY_ONLY
- MEMORY_ONLY_SER
A: MEMORY_ONLY: data is represented as JVM objects on which computation can be done directly.  Unfortunately, JVM objects are somewhat bloated (often using 2-5x more memory than the actual data).
MEMORY_ONLY_SER: data is seralized to a compact byte format.  This saves space, but data partitions must be transformed to JVM objects every time computation is done on them.

Q: Is data that is cached at the DISK_ONLY level serialized?
A: Yes, so it will actually use less storage space than MEMORY_ONLY (not serialized) uses of memory.

Q: What is df.cache() a shortcut for?
A: df.persist(StorageLevel.MEMORY_ONLY)

Q: What is the difference between these to Spark caching levels, and what are the tradeoffs?
- MEMORY_ONLY
- MEMORY_ONLY_2
A: MEMORY_ONLY only keeps the data in RAM on one worker.
MEMORY_ONLY_2 keeps it in RAM on two workers.  This uses more memory in the cluster (unfortunately), but it means we have more flexibility: when doing computation on an RDD partition, Spark has two workers with cached data it can choose from, so it can pick the worker that is less busy.
Also, if a worker with cached data crashes, we can use the other worker, avoiding a long delay to recompute the data that had been cached.

Q: What is the difference between normal partitioning and hash partitioning in Spark?
A: In normal partitioning, the rows are divided roughly evenly into partitions, and there is no guarantee about what row will be in what partition.
With hash partitioning, the user specifies a "key" (usually one or more columns).  For each row, we compute hash(key) % partition_count to get a partition number.  This means that a partition can contain rows with different keys, but all rows for the same key will end up in the same partition.

Q: What is the idea of partial aggregates in Spark?
A: For some aggregates (e.g., sum, count, avg), we can compute partial results prior to the exchange, often saving network I/O.
For example, instead of sending ("A", 1), ("A", 2), ("B", 4) when we're summing per key, a task could send this:
("A", 3), ("B", 4)
Of course, the receiving task might get "A" values from other executors too, so 3 is a partial sum, not the total sum.

Q: One optimization that is part of adaptive query execution (AQE) in Spark is "partition coallescing".  What is it?
A: For GROUP BY and similar queries, Spark needs to bring together related data with hash partitioning, using partition_num = hash(key) % partition_count, for each row.
partition_count could be anything, but needs to be the same for the entire Spark session.  It is 200 by default, which often creates many very small partitions (which is slow later).
The idea of partition coalescing is that the boss checks the sizes of fragments of partition data before those partitions are sent over the network.  It then combines some partitions together to form bigger partitions, ideally similar in size.  Executors are informed of the combining decisions.  So there will generally be <200 partitions in the end.

Q: What does bucketBy do when Spark is writing parquet files?
A: Each task that is writing data will bucket the data by key (this is basically the same as partitioning, but in Spark "partitioning" refers to memory/RDDs, and "bucketing" refers to files).  Data for each bucket is written to a different Parquet file.
This means that when we read the data back in, we can get an RDD that is already hash partitioned by that key.  If we want to do a GROUP BY (on the bucket key) followed by an aggregate, we can skip the network shuffle/exchange that would usually be necessary.

Q: What does SMJ stand for?
A: Shuffle Sort Merge Join

Q: What does BHJ stand for?
A: Broadcast Hash Join

Q: How does a distributed JOIN work with BHJ?
A: This is best for a join between a big table and small table.
The small table is loaded to a dictionary and sent to every machine that will work on a partition of the big table.
Tasks operating on partitions of the big table can independently loop over rows, matching keys with the dictionary.
The dictionary needs to fit in memory on every machine, so this is only a good strategy when the small table is small enough.

Q: How does a distributed JOIN work with SMJ?
A: Table A and table B are both hash partitioned on the column we are joining on.  Then we shuffle data to that we get pairs of partitions (one from A, one from B) with keys in common on the same executor.
The rows in these partitions are sorted by key, so tasks doing the JOIN can loop over the rows together, efficiently finding matches.