Estoy tratando de usarenfoque de transmisión estructurada usando Spark-Streaming basado en DataFrame / Dataset API para cargar un flujo de datos desde Kafka.

Yo suelo:

Spark 2.10Kafka 0.10chispa-sql-kafka-0-10

Spark Kafka DataSource ha definido un esquema subyacente:

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

Mis datos vienen en formato json y se almacenan en elvalor columna. Estoy buscando una forma de extraer el esquema subyacente de la columna de valor y actualizar el marco de datos recibido a las columnas almacenadas envalor? Intenté el siguiente enfoque pero no funciona:

 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()

Aquí estoy obteniendo la excepciónorg.apache.spark.sql.AnalysisException: Can't extract value from value#337; porque en el momento de la creación de la secuencia, no se conocen los valores internos ...

¿Tienes alguna sugerencia?