Volgens de fout heb je al een string, (je hebt al df.selectExpr("CAST(value AS STRING)") gedaan ), dus je moet proberen de Row-gebeurtenis als een String te krijgen , en niet een Array[Byte]
Begin met het veranderen van
val valueStr = new String(record.getAs[Array[Byte]]("value"))
naar
val valueStr = record.getAs[String]("value")
Ik begrijp dat je misschien al een cluster hebt om Spark-code uit te voeren, maar ik raad je aan om nog steeds te kijken naar de Kafka Connect Mongo Spoelbakaansluiting zodat u uw eigen Mongo-schrijver niet in Spark-code hoeft te schrijven en te onderhouden.
Of u kunt ook rechtstreeks Spark-datasets naar mongo schrijven