我试图从[Databricks] [1]重现该示例，并将其应用到Kafka的新连接器并进行火花结构化流式传输，但是我无法使用Spark中的现成方法正确解析JSON ...
注意：该主题以JSON格式写入Kafka。
val ds1 = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", IP + ":9092")
.option("zookeeper.connect", IP + ":2181")
.option("subscribe", TOPIC)
.option("startingOffsets", "earliest")
.option("max.poll.records", 10)
.option("failOnDataLoss", false)
.load()
以下代码将不起作用，我认为这是因为列json是字符串并且与from_json签名方法不匹配...
val df = ds1.select($"value" cast "string" as "json")
.select(from_json("json") as "data")
.select("data.*")
有小费吗？
[更新]示例工作：https : //github.com/katsou55/kafka-spark-structured-streaming-example/blob/master/src/main/scala-2.11/Main.scala
首先，您需要为JSON消息定义架构。例如
val schema = new StructType()
.add($"id".string)
.add($"name".string)
现在，您可以在from_json以下方法中使用此架构。
val df = ds1.select($"value" cast "string" as "json")
.select(from_json($"json", schema) as "data")
.select("data.*")
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句