如何在 Kafka connect 0.10 和 Spark Structured Streaming 中使用 from_json？

Question

我试图重现 [Databricks][1] 中的示例并将其应用于 Kafka 的新连接器和 spark 结构化流但是我无法使用 out-of-the 正确解析 JSON -Spark 中的框方法...

注意：主题以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

Answer 1

首先，您需要为 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.*")

如何在 Kafka connect 0.10 和 Spark Structured Streaming 中使用 from_json？

How to use from_json with Kafka connect 0.10 and Spark Structured Streaming?

scala

apache-kafka

apache-spark

apache-kafka-connect

spark-structured-streaming