異常在線程「主」org.apache.spark.sql.AnalysisException：

Question

我想我的手在kafka火花結構化流，但得到一些異常，如異常在線程「主」「org.apache.spark.sql.AnalysisException ：無法解析'device'給定的輸入列：[value，offset，partition，key，timestamp，timestampType，topic];

附上我的代碼

import org.apache.spark.sql._ 
import org.apache.spark.sql.functions._ 
import org.apache.spark.sql.types.StructType 
import org.apache.spark.sql.types._ 
import org.apache.spark.sql.streaming.ProcessingTime 
case class DeviceData(device: String, deviceType: String, signal: String) 

object dataset_kafka { 
    def main(args: Array[String]): Unit = { 
    val spark = SparkSession 
      .builder() 
      .appName("kafka-consumer") 
      .master("local[*]") 
      .getOrCreate() 
     import spark.implicits._ 

     spark.sparkContext.setLogLevel("WARN") 


    val df = spark 
     .readStream 
     .format("kafka") 
     .option("kafka.bootstrap.servers", "172.21.0.187:9093") 
     .option("subscribe", "test") 
     .option("startingOffsets", "earliest") 
     .load() 
     println(df.isStreaming) 
     println(df.printSchema()) 

    val ds: Dataset[DeviceData] = df.as[DeviceData] 

    val values = df.select("device").where("signal == Strong") 

    values.writeStream 
      .outputMode("append") 
      .format("console") 
      .start() 
      .awaitTermination() 


    } 
} 

任何幫助如何解決這個問題？

Answer 1

卡夫卡流總是產生以下字段：value，offset，partition，key，timestamp，timestampType，topic。在你的情況下，你對value感興趣，但請注意values are always deserialized as byte arrays，因此，在反序列化JSON之前需要強制轉換爲字符串。

試試下面的代碼：

import org.apache.spark.sql.functions.from_json 
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder 
import spark.implicits._ 

val kafkaStream = 
    spark.readStream 
    .format("kafka") 
    .option("kafka.bootstrap.servers", "172.21.0.187:9093") 
    .option("subscribe", "test") 
    .option("startingOffsets", "earliest") 
    .load() 

// If you don't want to build the schema manually 
val schema = ExpressionEncoder[DeviceData]().schema 

val ds = kafkaStream.select(from_json($"value".cast("string"), schema)).as[DeviceData] 

val values = ds.filter(_.signal == "Strong").map(_.device)