Как обработать объекты класса case Scala из Kafka с помощью потоковых запросов?

Question

Я использую интеграцию Kafka + Spark, где я отправляю объект класса дела (веб-сайт) и карту в искру.

    case class Website(id: Int, name: String)

    implicit val productSchema = Encoders.product[Website]
    val website = Website(1,"lokesh")
    EmbeddedKafka.publishToKafka(topic, website.toString)(config,new StringSerializer)

    val df:Dataset[Website] = spark
          .readStream
          .format("kafka")
          .option("subscribe", topic)
          .option("kafka.bootstrap.servers", "localhost:1244")
          .option("startingoffsets", "earliest")
          .load()
          .select("value") 
          .as[Website]

Я получаю сообщение об ошибке

Исключение в потоке "main" org.apache.spark.sql.AnalysisException: не удается разрешить 'id' для указанных входных столбцов: [значение];

Answer 1

tl; dr Используйте правильный формат сериализации, например, JSON или Avro.

---

Следующий код отправляет текстовое представление класса Website.

EmbeddedKafka.publishToKafka (topic, website.toString) (config, new StringSerializer)

Следующий код принимает текстовое представление как Array[Byte]:

.select ("value")

Так что вам будет лучше привести значение в строку и просто ... разобрать, чтобы разобрать объект, например, .select($"value" cast "string").

При этом вам будет лучше отправить JSON-представление объекта website, которое значительно упростит синтаксический анализ. Вы также можете использовать разделенный запятыми «формат сериализации», но для этого потребуется, чтобы у ваших website не было полей с запятыми.

---

AnalysisException: не удается разрешить 'id 'заданные входные столбцы: [значение]

Вышеуказанное исключение говорит о том, что вы хотите создать объект типа Website (который состоит из полей id и name) изvalue, что явно невозможно.

После .select("value") единственным столбцом является value (очевидно), и поэтому .as[Website] невозможно, так как нет "частей" типа Website длязаполнить (по имени).