[SCALA] 어떻게 구조화 스트리밍을 사용 카프카에서 JSON 형식의 기록을 읽어?

어떻게 구조화 스트리밍을 사용 카프카에서 JSON 형식의 기록을 읽어?

나는 카프카의 데이터 스트림을로드 할 수 DataFrame / 데이터 집합 API를 기반으로 스파크 - 스트리밍을 사용하여 구조화 된 스트리밍 방식을 사용하려합니다.

나는 사용한다:

스파크 카프카 데이터 소스는 기본 스키마를 정의하고있다 :

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

내 데이터는 JSON 형식으로 와서 그들은 값 컬럼에 저장됩니다. 나는 값에 저장 컬럼에 값 열 및 업데이트받은 dataframe에서 기본 스키마를 추출하는 방법을 방법을 찾고 있어요? 나는 아래의 방법을 시도했지만 작동하지 않습니다 :

 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을 얻고있다 : 값 # 337에서 값을 추출 할 수 없습니다; 때문에 스트림의 생성시에, 값은 내부 알려지지 않은 ...

의견 있으십니까?

해결법

1. ==============================

   1.스파크 관점 값으로부터 단지 바이트의 시퀀스이다. 그것은 직렬화 형식이나 내용에 대해 지식이 없습니다. 추출 할 수 있도록 당신이 먼저 구문 분석해야 제기했다.

   스파크 관점 값으로부터 단지 바이트의 시퀀스이다. 그것은 직렬화 형식이나 내용에 대해 지식이 없습니다. 추출 할 수 있도록 당신이 먼저 구문 분석해야 제기했다.

   데이터가 JSON 문자열로 직렬화하는 경우 두 가지 옵션이 있습니다. 당신은 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))
   

   또는 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: _*)
   

   원하는 유형 나중에 캐스팅.

from https://stackoverflow.com/questions/43297973/how-to-read-records-in-json-format-from-kafka-using-structured-streaming by cc-by-sa and MIT license