Apache Spark：Kafka以自定义格式编写

我正在构建一个使用Kafka主题的Spark SQL应用程序，转换一些数据，然后使用特定的JSON对象写回单独的Kafka主题。

现在我能够查询/转换我想要的内容并编写它：

Dataset reader = myData.getRecordCount();
reader.select(to_json(struct("record_count")).alias("value"))

.write()
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("topic", "new_separate_topic")
.save();

这会产生如下记录：

{
"record_count": 989
}
我需要的是，这一点JSON是一个更大的JSON对象的有效负载（子）属性，我们将其用作我们的微服务的标准消费者对象。

我想写的主题实际上是这样的：

{
"id": "ABC123",
"timestamp": "2018-11-16 20:40:26.108",
"user": "DEF456",
"type": "new_entity",
"data": {

  "record_count": 989
}

}
此外，“id”，“user”和“type”字段将从外部填充 - 它们将来自触发整个过程的原始Kafka消息。基本上，我需要为我想写入Kafka的元数据/对象注入一些值，并将“data”字段设置为上面的Spark SQL查询的结果。