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

Question

私はstructured streaming approachを使用して、DataFrame/Dataset APIに基づくSpark-Streamingを使用して、Kafkaからのデータストリームを読み込もうとしています。

Iが使用

：カフカDataSourceが、基礎となるスキーマを定義している

• スパーク2.10
• カフカ0.10
• 火花SQL-カフカ-0-10

スパーク：

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

私のデータはjson形式です値カラムに格納されます。値の列から基になるスキーマを抽出し、受信したデータフレームをの値に格納された列に更新する方法を探していますか？私は、以下のアプローチを試みたが、それは動作しません：

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;を取得していますあなたが何か提案はありますか？

Answer 1

Sparkの観点からは、valueは単なるバイトシーケンスです。シリアライズの形式や内容についての知識はありません。ファイルを抽出するには、まず解析する必要があります。

データがJSON文字列としてシリアル化される場合は、2つのオプションがあります。 、パスによる抽出フィールドはget_json_objectを用い

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にcast：あなたはStringTypeにcastvalueとfrom_jsonを使用してスキーマを提供することができる

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。