Как сериализовать строку в Scala

Question

У меня есть DStream [String, String]. Я использую foreachRDD для получения каждого RDD и публикую sh в Кафке. Проблема, с которой я столкнулся, заключается в том, что мне нужно гарантировать, что String сериализуется, а значение моего RDD не сериализуемо по неизвестной причине. Кафка ожидает получить StringSerializer в качестве значения, но, как вы можете видеть на изображении ниже, мой DStream не сериализовал String. Как я могу преобразовать String, не сериализуемый в serializabel до публикации sh в Kafka? Я мог бы изменить kafConf, но я бы предпочел изменить значение вместо конфигурации Kafka.

  def kafkaConf(brokers : String) = {
    val props = new HashMap[String, Object]()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    props
  }

Ошибка публикации в kafka

Answer 1

Ошибка ничего не говорит о строках. Dstream не сериализуем, то есть вы разделили его часть между областями исполнения, что заставляет Spark думать, что ему нужно сериализовать его

Вы действительно должны показать весь свой код, но для использования KafkaProducer вам нужна Spark Streaming. необходимо использовать foreachPartition, затем создать источник внутри этого блока.

Для каждого раздела вы l oop над каждым СДР, а затем используйте метод KafkaProducer.send

Вам не нужно беспокоиться о сериализации, если вы не хотите определить свой собственный

Answer 2

Я не могу сказать точное решение без кода. Я думаю, что ваша проблема не связана со свойствами Кафки.

В журнале ошибок Spark пытается сериализовать класс и произошел сбой.

Проверьте код внутри блока foreachRDD. Я думаю, что вы использовали несериализуемый класс. Проверьте свои классы и добавьте Serializable орудия к вашему классу, если можете. или просто попробуйте использовать тип String.