Title: CS544 / 08-spark
Tags: cs544-spark

Q: What is the difference between a data warehouse and data lake?
A: Both are a place where we collect data for analysis (OLAP workloads).
A warehouse is a database with integrated storage+computation.
A data lake is a combination of distributed storage (for example, HDFS) and decoupled analytics engine (for example, Spark)

Q: In MapReduce, what parameters do map and reduce functions take?
A: def map(key, value)
def reduce(key, values)
Note that with reduce, all values with the same key will go to a single reduce call on a single machine.  During map, different values with the same key go to different map calls, perhaps even on different machines.

Q: A reduce task might have the reduce function called several times.  What can we say about the keys across these multiple calls for a single task?
A: The keys will be sorted

Q: What determines the number of output files in a MapReduce job?
A: The number of reducers, because each reducer produces one output file.

Q: What are some performance advantages Spark has over MapReduce?
A: intermediate data is handled better (could directly pass from one computation to the next, without materializing in HDFS)
a chain of RDDs lets us specify many steps to get to a desired end result, but lazy execution means that an optimizer could find more efficient ways to get to that same result

Q: If you have a regular Python list, how can you convert it into a Spark RDD?
A: rdd = sc.parallelize(YOUR LIST)

Q: What is the relationship between actions, operations, and transformations?
A: actions and transformations are both kinds of operations
transformations are lazy (no work done yet), and describe how to produce an output RDD given an input RDD
actions actually materialize some output (on the screen, in an HDFS file, etc), triggering whatever work is necessary to compute that output.  An action is often at the end of a chain of transformations.

Q: A Spark task:
  - runs on a _________
  - operates on a _________
A: runs on a single CPU core
operates on a single partition, which is loaded entirely to memory

Q: Why might you want bigger Spark partitions?
A: Each partition corresponds to a task, and there is an overhead to starting tasks.  Bigger partitions => fewer partitions => less overhead.

Q: Why might you want smaller Spark partitions?
A: - use more cores that are available in the cluster
- balance work more evenly across executors
- use less memory at any given time

Q: What are two categories of Spark transformations?
A: narrow transformation: one input partition corresponds to one output partition
wide transformation: each output partition might need data from multiple input partitions (and vice versa)
if partitions are on different physical machines, wide transformations often cause network I/O (rows need to be shuffled/exchanged across the network)

Q: If df is a Spark DataFrame, how would we start/stop caching it?
A: df.cache() # start
df.unpersist() # stop

Q: What does RDD stand for, in the context of Spark?
A: Resilient Distributed Dataset

Q: How do you create a SparkData corresponding to a CSV at location PATH?
A: df = spark.read.format("csv").load(PATH)

Q: How can you see the number of RDD partitions for a Spark DataFrame df?
A: df.rdd.getNumPartitions()

Q: Say you run this:
spark.read.format("csv").load("test.csv")
And test.csv looks like this:
a,b
1,2
3,4
What will be the first row of data?
A: a,b
Because headers are off by default.  Add this:
spark.read.format("csv").option("header", True).load("test.csv")

Q: Say you run this:
spark.read.format("csv").option("header", True).load("test.csv")
And test.csv looks like this:
a,b
1,2
3,4
What will be the data types for the columns?
A: strings, because everything in a CSV is a string.
To get ints, use this:
spark.read.format("csv").option("header", True).option("inferSchema", True).load("test.csv")

Q: If you want to select the "x" column of a Spark DataFrame df, add 1 to it, and call the result column y, how would you do that?
A: df.select(expr("x+1").alias("y"))

Q: Spark gives an error about a file already existing when you run this:
df.select(columns).write.format("parquet").save(PATH)
What can you add to replace the previous file?
A: .mode("overwrite")
Like this:
df.select(columns).write.format("parquet").mode("overwrite").save(PATH)

Q: If df is a Spark DataFrame, what are two ways you can make it appear as a view in Hive (for the purpose of upcoming SQL queries)?
A: df.createTempView("VIEW_NAME")
df.createOrReplaceTempView("VIEW_NAME")
The former raises an exception if the view already exists.

Q: If you have a query, say "SELECT * FROM data", how can you run it with Spark SQL?  Assume your Spark session is in a variable "spark".
A: spark.sql("SELECT * FROM data")

Q: What type does spark.sql("QUERY") return?
A: a Spark DataFrame

Q: How do you write a Spark DataFrame (say df) as a Hive table?
A: df.write.saveAsTable("TABLE_NAME")

Q: What does spark.sql("SHOW TABLES") do?
A: Gives you a DataFrame listing all the tables AND all the views.  You'll need to do .show(), .toPandas(), or similar to actually see it.

Q: If tv is a table OR view in Hive, how can you get the corresponding Spark DataFrame?
A: spark.table("tv")

Q: What kind of Spark operation is .count()?
A: It depends.
In df.count() where df is a DataFrame, then it is an ACTION, because it actually triggers the work to get a number back.
In groups.count() where groups is a GroupedData, then it is a TRANSFORMATION, because the counts per group might only be an intermediate result that will be further processed.

Q: How are grouping and windowing similar and different?
A: In each case, we pull together related data into groups/partitions.
In grouping, there is one output row per group (e.g., an AVG stat).
In windowing, all the rows in the partitions can be part of the output, but they can have stats relative to the rest of the partition (e.g., a row_number() relative to the row's position with the partition).

Q: Say you want to filter the results of a Spark SQL query based on a value computed with a windowing function.  Why is this tricky, and what are some solutions?
A: We can't use WHERE -- that runs prior to the windowing function.  We can't use HAVING -- that is only for GROUP BY.
Options:
1. nested SQL query (the outer query can have a WHERE clause to filter)
2. add a where filter using the DataFrame API after the SQL query (this is an excellent case for mixing and matching Spark SQL with Spark DataFrames).
Example of option 2:
spark.sql("""
SELECT area, Call_Number, row_number() OVER (PARTITION BY area ORDER BY Call_Number ASC) AS num
FROM calls
""").where("num <= 3").toPandas()

Q: JOINs typically have ON clauses.  Say your ON clause is something like this:
ON tbl1.colA = tbl2.colB
What is the technical name for this kind of JOIN?
A: "equi join"

Q: If you only want to get the unique values in column C of table T, what SQL can you write?
A: SELECT DISTINCT C FROM T;

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.

Q: Is this deterministic?
df.randomSplit([0.75, 0.25], seed=544)
A: Not completely, because the deterministic split happens at the partition level.  If the number of partitions changes (e.g., because the number of workers changes and Spark decides a different partition count would be ideal), then the overall split won't be deterministic.

Q: Consider Spark's DecisionTreeRegressionModel and DecisionTreeRegressor.  Which one is fitted?
A: DecisionTreeRegressionModel is fitted.
DecisionTreeRegressor is NOT fitted.
Usually, the class names ending in "Model" represent fitted models in Spark.

Q: dt = DecisionTreeRegressor(featuresCol="????", labelCol="????")
We can only pass in one featuresCol -- how do we handle multiple features?
A: Use a vector assembler to transform a DataFrame we want to train/predict on, adding a column of vectors that contains data from multiple columns:
va = VectorAssembler(inputCols=["col1", "col2", ...], outputCol="features")

Q: If we have a fitted model in Spark, how do we used it to make predictions?
A: df_WITH_predictions = model.transform(df_WITHOUT_predictions)
.transform(...) in Spark is used both for preprocessing AND prediction.

Q: A Spark "Pipeline" object is an unfit model, consisting of multiple transformers and ending in an estimator.  When we call "fit" on it, we get a fitted PipelineModel.  What does it contain?
A: All the transformers of Pipeline are copied to PipelineModel.
"fit" is called on the estimator at the end of Pipeline, and this "fit" returns a transformer.  The transformer is copied to the end of PipelineModel, such that PipelineModel is a series of transformers.

Q: The traditional decision tree training algorithm considers splits on every column, for every possible split between rows (a threshold on a column is a split).  Where does Spark consider splits?
A: It still looks at every column, but we configure a number of bins on each column, for example: dt.setMaxBins(N).  A histogram is computed on this with N buckets, and the thresholds considered are those that split the buckets (so N-1 possible split points per column).

Q: What is an "equi-depth" histogram?
A: Each bucket contains a similar number of data points, even if that means some buckets cover a wider range of values than others (e.g., 1-3, 3-10, 10-100, etc).
In contrast, most histograms have buckets with fixed intervals (e.g., 0-10, 10-20, 20-30, etc).