Kafka ile from_json 0.10 ve Spark Structured Streaming nasıl kullanılır?

Question

Örneği [Databricks] [1] 'den yeniden üretmeye çalışıyorum ve bunu yeni bağlantıya kafka ve kıvılcım yapılandırılmış akışa uyguladım, ancak kıvılcımdaki kutudan çıkma yöntemlerini kullanarak jsonu doğru bir şekilde ayrışamıyorum.

not: konu kafon'a json biçiminde yazılmıştır.

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

Aşağıdaki kod çalışmaz, ben

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

Herhangi bir ipucu ... kolon json bir dize ve yöntem from_json imzası eşleşmiyor çünkü inanıyoruz?

[GÜNCELLEME] Örnek çalışma: https://github.com/katsou55/kafka-spark-structured-streaming-example/blob/master/src/main/scala-2.11/Main.scala

Answer 1

Öncelikle JSON mesajı için şema tanımlamak gerekir. Örneğin

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

Şimdi aşağıda gibi from_json yönteminde bu şemayı kullanabilirsiniz.

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