kafka connect 0.10とSpark Structured Streamingでfrom_jsonを使用するには？

Question

私は[Databricks] [1]からサンプルを再現し、kafkaの新しいコネクタに適用し、構造化されたストリーミングを実行しようとしていましたが、sparkのout-of-the-boxメソッドを使ってjsonを正しく解析できません。

注：トピックはjson形式でkafkaに書き込まれます。

val ds1 = spark 
      .readStream 
      .format("kafka") 
      .option("kafka.bootstrap.servers", IP + ":9092") 
      .option("zookeeper.connect", IP + ":2181") 
      .option("subscribe", TOPIC) 
      .option("startingOffsets", "earliest") 
      .option("max.poll.records", 10) 
      .option("failOnDataLoss", false) 
      .load() 

次のコードは、私が

val df = ds1.select($"value" cast "string" as "json") 
       .select(from_json("json") as "data") 
       .select("data.*") 

任意のヒント...列JSONが文字列であると署名from_json方法と一致していないので、それはだと信じて、動作しないのだろうか？

[UPDATE]例の作業： https://github.com/katsou55/kafka-spark-structured-streaming-example/blob/master/src/main/scala-2.11/Main.scala

Answer 1

は、まず、あなたのJSONメッセージのスキーマを定義する必要があります。例えば、

val schema = new StructType() 
    .add($"id".string) 
    .add($"name".string) 

このスキーマは、以下のようにfrom_jsonメソッドで使用できます。

val df = ds1.select($"value" cast "string" as "json") 
      .select(from_json($"json", schema) as "data") 
      .select("data.*")