Reading from Kafka with Spark in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output schema of this Spark Kafka read?

Given the following Spark code to read from Kafka, what is the schema of the resulting DataFrame?

Apache Spark

df = spark.read.format("kafka")\
  .option("kafka.bootstrap.servers", "localhost:9092")\
  .option("subscribe", "topic1")\
  .load()
df.printSchema()

A root\n |-- key: binary (nullable = true)\n |-- value: binary (nullable = true)\n |-- topic: string (nullable = true)\n |-- partition: integer (nullable = true)\n |-- offset: long (nullable = true)\n |-- timestamp: timestamp (nullable = true)\n |-- timestampType: integer (nullable = true)

B root\n |-- key: binary (nullable = false)\n |-- value: binary (nullable = false)\n |-- topic: string (nullable = false)\n |-- partition: integer (nullable = false)\n |-- offset: long (nullable = false)\n |-- timestamp: timestamp (nullable = false)\n |-- timestampType: integer (nullable = false)

C root\n |-- key: string (nullable = true)\n |-- value: string (nullable = true)\n |-- topic: string (nullable = true)\n |-- partition: string (nullable = true)\n |-- offset: string (nullable = true)\n |-- timestamp: string (nullable = true)\n |-- timestampType: string (nullable = true)

D root\n |-- key: string (nullable = false)\n |-- value: string (nullable = false)\n |-- topic: string (nullable = false)\n |-- partition: integer (nullable = true)\n |-- offset: long (nullable = true)\n |-- timestamp: timestamp (nullable = true)\n |-- timestampType: integer (nullable = true)

💡 Hint

Kafka message keys and values are binary by default in Spark.

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

❓ data_output

intermediate

What is the count of messages read from Kafka?

Assuming the Kafka topic 'topic1' has 100 messages, what will be the output of this code?

Apache Spark

df = spark.read.format("kafka")\
  .option("kafka.bootstrap.servers", "localhost:9092")\
  .option("subscribe", "topic1")\
  .load()
count = df.count()
print(count)

A 0

B 100

C None

D Raises an error

💡 Hint

Spark reads all available messages in the topic partition(s) when loading.

The count() action returns the number of rows in the DataFrame, which corresponds to the number of messages read from Kafka. Since the topic has 100 messages, the count is 100.

🔧 Debug

advanced

Why does this Spark Kafka read code fail with a timeout?

This code snippet to read from Kafka fails with a timeout error. What is the most likely cause?

Apache Spark

df = spark.read.format("kafka")\
  .option("kafka.bootstrap.servers", "wronghost:9092")\
  .option("subscribe", "topic1")\
  .load()

A The Kafka bootstrap server address is incorrect or unreachable.

B The topic name 'topic1' does not exist in Kafka.

C Spark session is not initialized before reading.

D The format 'kafka' is not supported by Spark.

💡 Hint

Timeout errors usually relate to network or server connection issues.

A timeout error when reading from Kafka usually means Spark cannot connect to the Kafka bootstrap server. An incorrect or unreachable server address causes this.

🚀 Application

advanced

How to convert Kafka message values to strings in Spark?

You read Kafka messages with Spark and want to convert the 'value' column from binary to string. Which code snippet correctly does this?

Apache Spark

from pyspark.sql.functions import col, expr

# df is the DataFrame read from Kafka
converted_df = ???

A df.select(col("value").toString().alias("value_str"))

B df.selectExpr("CAST(value AS STRING) as value_str")

C df.selectExpr("value.toString() as value_str")

D df.select(col("value").cast("string").alias("value_str"))

💡 Hint

Use Spark SQL functions or DataFrame API to cast binary to string.

Casting the 'value' column from binary to string using col().cast('string') is the correct way. selectExpr with CAST(value AS STRING) works but is less flexible. value.toString() is not a valid Spark SQL expression.

🧠 Conceptual

expert

What happens if you use 'startingOffsets' option with 'latest' in Spark Kafka read?

When reading from Kafka with Spark, setting option startingOffsets to 'latest' means:

A Spark reads messages from a random offset in the topic.

B Spark reads all messages from the beginning of the topic.

C Spark reads only new messages arriving after the query starts.

D Spark reads messages from the earliest offset available.

💡 Hint

Think about what 'latest' means in Kafka offset context.

Setting startingOffsets to 'latest' tells Spark to start reading only new messages that arrive after the streaming query starts, ignoring older messages.

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