如何使用from_json與kafka connect 0.10和Spark結構化流？

我試圖從[Databricks] [1]重現這個例子，並將它應用到kafka的新連接器，並引發了結構化流式處理，但是我無法正確使用spark中的開箱即用方法來解析json。如何使用from_json與kafka connect 0.10和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

來源

2017-02-28 carlos rodrigues

首先，您需要定義架構您的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.*")

來源

2017-02-28 17:18:06 abaghel

如果你有編譯器警告「價值$不是成員......」請不要忘了進口spark.implicits._我花了額外的5-10分鐘找出 – user1459144

我的問題是，哪個庫提供了稱爲「from_json」的函數？我似乎不能放置！請幫助.. – Gyan

@Raghav - > import org.apache.spark.sql.functions._ 查看這裏的例子：https：//github.com/katsou55/kafka-spark-structured-streaming-example/blob/主/ SRC /主/階-2.11/Main.scala –

如何使用from_json與kafka connect 0.10和Spark結構化流？

回答

相關問題