Spark structured streaming - how to queue bytes value to Kafka?

Question

I'm writing a Spark application that uses structured streaming. The app reads messages from a Kafka topic topic1, constructs a new message, serializes it to an Array[Byte] and publishes them to another Kafka topic topic2.

The serializing to a byte array is important because I use a specific serializer/deserializer that the downstream consumer of topic2 also uses.

I've trouble producing to Kafka though. I'm not even sure how to do so..there's only plenty of examples online about queueing JSON data.

The code -

case class OutputMessage(id: String, bytes: Array[Byte])

implicit val encoder: Encoder[OutputMessage] = org.apache.spark.sql.Encoders.kryo

val outputMessagesDataSet: DataSet[OutputMessage] = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "server1")
  .option("subscribe", "topic1")
  .load()
  .select($"value")
  .mapPartitions{r =>
     val messages: Iterator[OutputMessage] = createMessages(r)
     messages
  }

outputMessagesDataSet
  .writeStream
  .selectExpr("CAST(id AS String) AS key", "bytes AS value")
  .format("kafka")
  .option("kafka.bootstrap.servers", "server1")
  .option("topic", "topic2")
  .option("checkpointLocation", loc)
  .trigger(trigger)
  .start
  .awaitTermination

However, that throws exception org.apache.spark.sql.AnalysisException: cannot resolve 'id' given input columns: [value]; line 1 pos 5;

How do I queue to Kafka with id as the key and bytes as the value?

Answer 1

You can check the schema of the dataframe that "collects" the message. As you are collecting only the "value" field, incoming events arrive in the following form:

    +-------------------+
    | value             |
    +-------------------+
    | field1,field2,..  |
    +-------------------+
  

Yo need to query for the key as well like in the Spark documentation:

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

or

df.select(col("key").cast(StringType), col("value").cast(StringType))

Answer 2

As @EmiCareOfCell44 suggested, I printed out the schema -

If I do messagesDataSet.printSchema() then I get only one value with binary type. But if I do

val df = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "server1")
.option("subscribe", "topic1")
.load()

df.printSchema()

Then it prints

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

But the Dataframe hasn't undergone the transformation that is needed, which is done in

.mapPartitions{r =>
 val messages: Iterator[OutputMessage] = createMessages(r)
 messages
}

It looks like the Dataset's value has only one binary value.

I searched for some answers here, then found this post - Value Type is binary after Spark Dataset mapGroups operation even return a String in the function

I had an Encoder set up -

implicit val encoder: Encoder[OutputMessage] = org.apache.spark.sql.Encoders.kryo

That was causing the value to be converted into binary. Since OutputMessage is a scala class, the Encoder isn't required, so I removed it. After that, printing out the schema showed two fields (String and bytes which is what I wanted). After that, line .selectExpr("CAST(id AS String) AS key", "bytes AS value") worked perfectly well.