Title: 544
Tags: lec33

Q: How can you use spark.read in Spark to get rows corresponding to Kafka messages?
A: Use .format to specify data is coming from Kafka.  Use .option to specify the bootstrap servers for the Kafka cluster and the topic(s) to subscribe to.

Q: spark.read.???? returns a regular DataFrame.  How do you get a streaming DataFrame?
A: replace "read" with "readStream"

Q: When you read from a Kafka topic from Spark, what columns will your DataFrame have?
A: key, value, some others.  They columns you get will always be the same, regardless of the message format!
This usually means Spark has to do some extra work to extract different values out of the value, which probably contains some data structure serialized as bytes.

Q: Why is .config("spark.sql.shuffle.partitions", ????) important for Spark streaming, but not other Spark work?
A: The default number of shuffle partitions is 200, meaning that after group by or similar, there will be 200 partitions in the resulting RDD.  There is a 1-to-1 correspondence between tasks and partitions, so this means that further computation will (by default) cause hundreds of tasks to run -- the total startup costs of these tasks becomes a significant overhead.
In recent versions of Spark, a "partition coalescing" feature has been added that combines together many small partitions, usually reducing the results to far fewer than 200 partitions.  But the partition coalescing optimization is not supported for streaming queries, so Spark programmers need to choose this config carefully for streaming scenarios.

Q: What does a streaming Spark query do?
A: Moves data from a source to a sink, possibly with some transformations in between.

Q: How are source and sink specified in a Spark streaming query?
A: With readStream and writeStream, respectively:
q = spark.readStream(????).????.writeStream(????).start()

Q: When you uses Spark's from_json to deserialize to a struct/dictionary, you must specify a schema.  What if the schema lists a column not found in the dictionary?  What if the dictionary has a column not listed in the schema?
A: Schema has extra column: fill with NULL values
Dictionary has extra column: ignore it