Deserializing Protobuf from Kafka in Spark Structured Streaming

I'm testing an implementation at work that will see 300 million messages/day coming through, with plans to scale up enormously. There's one step that seems janky at the moment and I'd appreciate some advice.

I did take a stab at this with https://scalapb.github.io/sparksql.html but couldn't seem to get it to work, even following their maven instructions.

Currently I have a protobuf and a case class for the same model:

message MyThing { // proto
    required string id = 1;
}

case class MyThing(id: String)

Then I have a spark readStream

val df =  
  spark.readStream
    .format("kafka")
    // etc
    .load()

The kafka payload is in the "value" column, which is an Array[Byte] from the protobuf that was transmitted. I want to turn that binary column into Rows with a specific StructType.

What I have right now uses a weird syntax involving the case class:

val encoder = Encoder.product[MyThing]

df
  .select("value")
  .map { row => 
     // from memory so might be slightly off
     val proto = MyThingProto.parseFrom(row.getBinary(0)) 
     val myThing = MyThing.fromProto(proto)
     myThing
  }(encoder)
  .toDF()
  // business logic
  .writeStream
  ...//output

Can I make this more efficient/faster? The overhead involved in creating the case class seems excessive. I'd prefer to be able to do something like this:

  .map { row => 
     // from memory so might be slightly off
     val proto = MyThingProto.parseFrom(row.getBinary(0)) 
     val row = buildRow(proto)
     row
  }(encoder??) // what kind of encoder is used here?

  def buildRow(proto: MyThingProto): Row = 
      Row(proto.getId)

Would this be better? Or perhaps a UDF that uses the Kafka deserializer interface?

Thanks in advance.