7
我試圖從[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.*")
任何提示嗎?
如果你有編譯器警告「價值$不是成員......」請不要忘了進口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 –