如何在Kafka connect 0.10和Spark结构化流中使用from_json? [英] How to use from_json with Kafka connect 0.10 and Spark Structured Streaming?

问题描述

我试图重现[Databricks] [1]中的示例，并将其应用于Kafka的新连接器并进行Spark结构化流式传输，但是我无法使用Spark中的现成方法正确解析JSON. ..

I was trying to reproduce the example from [Databricks][1] and apply it to the new connector to Kafka and spark structured streaming however I cannot parse the JSON correctly using the out-of-the-box methods in Spark...

注意:该主题以JSON格式写入Kafka.

note: the topic is written into Kafka in JSON format.

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签名的方法不匹配...

The following code won't work, I believe that's because the column json is a string and does not match the method from_json signature...

    val df = ds1.select($"value" cast "string" as "json")
                .select(from_json("json") as "data")
                .select("data.*")

有什么提示吗?

[UPDATE]示例工作: https://github.com/katsou55/kafka-spark-structured-streaming-example/blob/master/src/main/scala-2.11/Main.scala

[UPDATE] Example working: https://github.com/katsou55/kafka-spark-structured-streaming-example/blob/master/src/main/scala-2.11/Main.scala

推荐答案

首先，您需要为JSON消息定义架构.例如

First you need to define the schema for your JSON message. For example

val schema = new StructType()
  .add($"id".string)
  .add($"name".string)

现在，您可以在from_json方法中使用此架构，如下所示.

Now you can use this schema in from_json method like below.

val df = ds1.select($"value" cast "string" as "json")
            .select(from_json($"json", schema) as "data")
            .select("data.*")

这篇关于如何在Kafka connect 0.10和Spark结构化流中使用from_json?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！