Reading from Kafka with Spark in Apache Spark - Cheat Sheet & Quick Revision

Recall & Review

beginner

What is Apache Kafka used for in data processing?

Apache Kafka is a system that lets you send and receive streams of data in real time. It works like a messaging service where data is published and consumed continuously.

beginner

How does Spark connect to Kafka to read data?

Spark uses a special connector called the Kafka source. It reads data from Kafka topics as a stream or batch by specifying Kafka servers and topics in the Spark read options.

beginner

What is the role of 'subscribe' option when reading from Kafka in Spark?

The 'subscribe' option tells Spark which Kafka topic(s) to listen to for incoming data. You can list one or more topics separated by commas.

intermediate

What data format does Spark receive from Kafka by default?

Spark receives Kafka data as binary key and value columns. You often need to convert or decode these bytes into readable strings or structured data.

intermediate

Why is it important to set 'startingOffsets' when reading from Kafka with Spark?

The 'startingOffsets' option controls where Spark starts reading data in Kafka. For example, 'earliest' reads from the oldest data, and 'latest' reads only new data. This helps control what data you process.

Which Spark method is used to read data from Kafka?

A spark.read.format("kafka")

B spark.read.kafka()

C spark.readStream.kafka()

D spark.kafka.read()

To read from Kafka, you use spark.read.format("kafka") for batch or spark.readStream.format("kafka") for streaming.

What does the 'subscribe' option specify in Spark Kafka reading?

A Data format

B Kafka broker addresses

C Starting offset position

D Kafka topic names

The 'subscribe' option tells Spark which Kafka topic(s) to read from.

What data type are Kafka messages received as in Spark by default?

A String

B Binary

C Integer

D JSON

Kafka messages come as binary data in Spark and need decoding.

Which option controls where Spark starts reading Kafka data?

A subscribe

B kafka.bootstrap.servers

C startingOffsets

D group.id

'startingOffsets' sets the position to start reading from Kafka.

To read Kafka data as a continuous stream in Spark, which method is used?

A spark.readStream.format("kafka")

B spark.read.format("kafka")

C spark.kafka.stream()

D spark.stream.read()

spark.readStream.format("kafka") reads Kafka data as a streaming source.

Explain how to set up Spark to read data from a Kafka topic.

Think about the options needed to connect and select data from Kafka.

Describe how Kafka data appears in Spark and what you need to do to use it.

Consider the data type Spark receives and how to make it readable.

Practice

1. What is the main purpose of using spark.readStream.format("kafka") in Apache Spark?

easy

A. To read live streaming data from Kafka topics

B. To write data into Kafka topics

C. To stop Kafka streaming jobs

D. To configure Kafka brokers

Solution

1. Step 1: Understand the function of spark.readStream.format("kafka")

   This command sets up Spark to read streaming data from Kafka, meaning it connects Spark to Kafka to get live data.

2. Step 2: Differentiate reading from writing or configuring

   Writing data or stopping jobs are different operations. This command is specifically for reading data streams.

3. Final Answer:

   To read live streaming data from Kafka topics -> Option A

4. Quick Check:

   Reading Kafka streams = To read live streaming data from Kafka topics [OK]

Hint: Remember: readStream + kafka = reading live data [OK]

Common Mistakes:

• Confusing reading with writing data
• Thinking it configures Kafka brokers
• Assuming it stops streaming jobs

2. Which of the following is the correct way to specify Kafka bootstrap servers when reading from Kafka in Spark?

easy

A. option("servers", "localhost:9092")

B. option("kafka.bootstrap.servers", "localhost:9092")

C. option("kafka.servers", "localhost:9092")

D. option("bootstrap.servers", "localhost:9092")

Solution

1. Step 1: Recall the exact option key for Kafka bootstrap servers

   The correct key is "bootstrap.servers" without the "kafka." prefix when used in Spark's option method.

2. Step 2: Compare all options

   Only option("bootstrap.servers", "localhost:9092") uses "bootstrap.servers" correctly. Others use incorrect keys with unnecessary prefixes.

3. Final Answer:

   option("bootstrap.servers", "localhost:9092") -> Option D

4. Quick Check:

   Correct Kafka server option key = option("bootstrap.servers", "localhost:9092") [OK]

Hint: Use exact key 'bootstrap.servers' for Kafka servers [OK]

Common Mistakes:

• Adding 'kafka.' prefix to option keys
• Using incorrect option names like 'kafka.servers'
• Omitting required option keys

3. Given the following Spark code snippet reading from Kafka, what will be the schema of the resulting DataFrame?

df = spark.readStream.format("kafka")\
  .option("kafka.bootstrap.servers", "localhost:9092")\
  .option("subscribe", "test-topic")\
  .load()

medium

A. DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int)

B. DataFrame with columns: key (string), value (string)

C. DataFrame with columns: message (string), offset (int)

D. DataFrame with columns: topic (string), message (string), timestamp (string)

Solution

1. Step 1: Understand default Kafka DataFrame schema in Spark

   When reading from Kafka, Spark returns a DataFrame with columns including key and value as binary types, plus metadata columns like topic, partition, offset, timestamp, and timestampType.

2. Step 2: Match schema with options

   DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int) correctly lists all these columns with their types. Other options simplify or rename columns incorrectly.

3. Final Answer:

   DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int) -> Option A

4. Quick Check:

   Kafka DataFrame schema = DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int) [OK]

Hint: Kafka data columns include key and value as binary by default [OK]

Common Mistakes:

• Assuming key and value are strings by default
• Ignoring metadata columns like offset and timestamp
• Confusing column names and types

4. Identify the error in the following Spark code reading from Kafka:

df = spark.readStream.format("kafka")\
  .option("kafka.bootstrap.servers", "localhost:9092")\
  .option("topic", "my-topic")\
  .load()

medium

A. The format should be 'kafkaStream' instead of 'kafka'

B. The option key for topic should be 'subscribe', not 'topic'

C. The bootstrap.servers option is misspelled

D. The load() method is missing parentheses

Solution

1. Step 1: Check the option keys for Kafka source in Spark

   The correct option key to specify Kafka topics to read is 'subscribe', not 'topic'.

2. Step 2: Verify other parts of the code

   Bootstrap servers key is correct, format 'kafka' is correct, and load() has parentheses.

3. Final Answer:

   The option key for topic should be 'subscribe', not 'topic' -> Option B

4. Quick Check:

   Use 'subscribe' to specify topics [OK]

Hint: Use 'subscribe' option for topics, not 'topic' [OK]

Common Mistakes:

• Using 'topic' instead of 'subscribe'
• Misspelling option keys
• Confusing format names

5. You want to read messages from Kafka in Spark and convert the binary 'value' column to a string column named 'message'. Which code snippet correctly achieves this?

hard

A. df.withColumn("message", df.value.toString())

B. df.select("value").cast("string").alias("message")

C. df.selectExpr("CAST(value AS STRING) as message")

D. df.select(df.value).withColumnRenamed("value", "message")

Solution

1. Step 1: Understand how to cast binary to string in Spark DataFrame

   The correct way is to use selectExpr with SQL expression: CAST(value AS STRING) and alias it as 'message'.

2. Step 2: Evaluate each option

   df.selectExpr("CAST(value AS STRING) as message") uses selectExpr correctly. df.select("value").cast("string").alias("message") tries to cast after select which is invalid. df.withColumn("message", df.value.toString()) uses toString() which converts to Python string, not Spark column. df.select(df.value).withColumnRenamed("value", "message") renames without casting to string.

3. Final Answer:

   df.selectExpr("CAST(value AS STRING) as message") -> Option C

4. Quick Check:

   Use selectExpr with CAST for binary to string [OK]

Hint: Use selectExpr("CAST(value AS STRING) as message") to convert [OK]

Common Mistakes:

• Using toString() on Spark columns
• Casting after select without aliasing
• Not renaming the column after casting