如何将 from_json 与 Kafka connect 0.10 和 Spark Structured Streaming 一起使用？

Posted 2023-03-10

【中文标题】如何将 from_json 与 Kafka connect 0.10 和 Spark Structured Streaming 一起使用？

【英文标题】：How to use from_json with Kafka connect 0.10 and Spark Structured Streaming?

【发布时间】：2017-07-19 07:24:53

【问题描述】：

我试图重现 [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

【参考方案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.*")

【讨论】：

如果您有编译器警告“值 $ 不是成员...”请不要忘记 import spark.implicits._ 我又花了 5-10 分钟才弄清楚

对我来说问题是，哪个库提供了名为“from_json”的函数？我似乎无法放置它！请帮忙..

@Raghav -> import org.apache.spark.sql.functions._ 在此处查看示例：github.com/katsou55/kafka-spark-structured-streaming-example/…

@abaghel - 我在这里指的是你的博客 - programcreek.com/java-api-examples/… 没有 java bean，我们不能解析/推断 json 数据吗？

@abaghel 我看不到控制台输出，即使它打印 isStreaming 为 true