Как использовать from_json с Kafka connect 0.10 и Spark Structured Streaming?

Я пытался воспроизвести пример из [Databricks] [1] и применить его к новому коннектору для Kafka и структурированной потоковой передачи, однако я не могу правильно проанализировать JSON, используя готовые методы в Spark...

примечание: тема написана на Kafka в формате JSON.

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

Следующий код не будет работать, я считаю, что столбец json является строкой и не соответствует методу from_json signature...

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

Какие-нибудь советы?

[ОБНОВЛЕНИЕ] Пример работы:https://github.com/katsou55/kafka-spark-structured-streaming-example/blob/master/src/main/scala-2.11/Main.scala

Ответ 1