scala - 構造化ストリーミングを使用して Kafka から JSON 形式のレコードを読み取る方法は?

Question

DataFrame/Dataset API に基づく Spark-Streaming を使用して、構造化されたストリーミング アプローチを使用して、Kafka からデータのストリームをロードしようとしています。

私が使う：

• スパーク 2.10
• カフカ 0.10
• spark-sql-kafka-0-10

Spark Kafka DataSource は、基礎となるスキーマを定義しています。

|key|value|topic|partition|offset|timestamp|timestampType|

私のデータはjson形式で提供され、値列に保存されます。基になるスキーマを value 列から抽出し、受信したデータフレームをvalueに格納された列に更新する方法を探していますか? 以下のアプローチを試しましたが、うまくいきません。

 val columns = Array("column1", "column2") // column names
 val rawKafkaDF = sparkSession.sqlContext.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers","localhost:9092")
  .option("subscribe",topic)
  .load()
  val columnsToSelect = columns.map( x => new Column("value." + x))
  val kafkaDF = rawKafkaDF.select(columnsToSelect:_*)

  // some analytics using stream dataframe kafkaDF

  val query = kafkaDF.writeStream.format("console").start()
  query.awaitTermination()

org.apache.spark.sql.AnalysisException: Can't extract value from value#337;ここで、ストリームの作成時に内部の値が不明であるため、例外が発生しています...

何か提案はありますか？