Reading from Kafka with Spark in Apache Spark - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to read data from Kafka using Spark.

Apache Spark

df = spark.readStream.format([1]).option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "topic1").load()

Drag options to blanks, or click blank then click option'

A"kafka"

B"json"

C"csv"

D"parquet"

Attempts:

3 left

💡 Hint

Common Mistakes

Using formats like "json" or "csv" instead of "kafka".

Forgetting to specify the Kafka bootstrap servers.

To read from Kafka, the format must be set to "kafka" in Spark.

2fill in blank

medium

Complete the code to select the Kafka message value as a string.

Apache Spark

messages = df.selectExpr("CAST(value AS [1]) AS value")

Drag options to blanks, or click blank then click option'

A"string"

B"boolean"

C"float"

D"int"

Attempts:

3 left

💡 Hint

Common Mistakes

Casting to numeric types like "int" or "float" without conversion.

Not casting the value at all.

Kafka message values are binary, so casting to "string" makes them readable text.

3fill in blank

hard

Fix the error in the code to write streaming data back to Kafka.

Apache Spark

query = messages.writeStream.format([1]).option("kafka.bootstrap.servers", "localhost:9092").option("topic", "output_topic").start()

Drag options to blanks, or click blank then click option'

A"json"

B"console"

C"kafka"

D"parquet"

Attempts:

3 left

💡 Hint

Common Mistakes

Using "console" or "json" as the sink format when writing to Kafka.

Forgetting to specify the Kafka bootstrap servers.

To write streaming data to Kafka, the format must be "kafka".

4fill in blank

hard

Fill both blanks to filter messages with value length greater than 5.

Apache Spark

filtered = messages.filter(length([1]) [2] 5)

Drag options to blanks, or click blank then click option'

A"value"

B>

C<

D"key"

Attempts:

3 left

💡 Hint

Common Mistakes

Filtering on "key" instead of "value".

Using the wrong comparison operator.

We filter on the "value" column where its length is greater than 5.

5fill in blank

hard

Fill all three blanks to create a dictionary of word counts from Kafka messages.

Apache Spark

word_counts = {word: [1] for word in words if [2](word) [3] 3}

Drag options to blanks, or click blank then click option'

Alen(word)

Blen

C>

Dcount

Attempts:

3 left

💡 Hint

Common Mistakes

Using the wrong variable in the condition.

Using incorrect comparison operators.

We count the length of each word and include only words longer than 3 characters.

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