Как читать записи в формате JSON из Kafka с помощью структурированной потоковой передачи?

Я пытаюсь использовать подход структурированной потоковой передачи с использованием Spark- Потоковая передача на основе DataFrame / Dataset API для загрузки потока данных из Kafka.

Я использую:

  • Искра 2.10
  • Кафка 0,10
  • искра-sql-кафка-0-10

Источник данных Spark Kafka определил базовую схему:

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

Мои данные поступают в формате json и хранятся в столбце значение. Я ищу способ извлечь базовую схему из столбца значений и обновить полученный фрейм данных до столбцов, хранящихся в 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()

Здесь я получаю Exception org.apache.spark.sql.AnalysisException: Can't extract value from value#337;, потому что во время создания потока значения внутри неизвестны ...

Есть ли у вас какие-либо предложения?


apache-kafka apache-spark apache-spark-sql spark-structured-streaming scala
person Stefan Repcek    schedule 08.04.2017    source источник

Ответы (1)


arrow_upward
13
arrow_downward

С точки зрения Spark value - это просто последовательность байтов. Он ничего не знает о формате или содержимом сериализации. Чтобы иметь возможность извлечь файл, вы должны сначала его проанализировать.

Если данные сериализуются в виде строки JSON, у вас есть два варианта. Вы можете cast value до StringType и использовать from_json и предоставить схему:

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions.from_json

val schema: StructType = StructType(Seq(
  StructField("column1", ???),
  StructField("column2", ???)
))

rawKafkaDF.select(from_json($"value".cast(StringType), schema))

или от cast до StringType, извлекать поля по пути, используя get_json_object:

import org.apache.spark.sql.functions.get_json_object

val columns: Seq[String] = ???

val exprs = columns.map(c => get_json_object($"value", s"$$.$c"))

rawKafkaDF.select(exprs: _*)

и cast позже к желаемым типам.

person zero323    schedule 08.04.2017
comment
FWIW, я считаю обе альтернативы неуклюжими: + альтернатива схемы требует спецификации схемы с использованием собственного кода Spark. Поэтому, если мы отправляем сложную схему, код должен проанализировать файл схемы (xsd и т. Д.) И построить этот объект. + get_json_object альтернативные заставляют индивидуально выдвигать пути / поля. Я не уверен, что это за снижение производительности. Я бы предпочел, чтобы Apache Spark поставлялся с более простым способом чистого приема файла схемы и создания объекта схемы Spark / Catalyst. - person Venki; 08.04.2019
comment
другой способ получить схему для сообщений (это еще один неуклюжий способ): stackoverflow.com/questions/48361177/ - person Venki; 08.04.2019