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.