Искра генерирует исключение не сериализуемого внутри операции foreachRDD - PullRequest
Новая
       17

Искра генерирует исключение не сериализуемого внутри операции foreachRDD

1 голос
/ 30 июня 2019

Я пытаюсь реализовать шаблон наблюдателя, используя потоки scala и spark.идея заключается в том, что всякий раз, когда я получаю запись из потока (от kafka), я уведомляю наблюдателя, вызывая метод "notifyObservers" внутри замыкания.Вот код:

поток предоставляется утилитами kafka.метод notifyObserver определен в абстрактный класс, следуя правилам шаблона.ошибка, я думаю, связана с тем, что методы не могут быть сериализованы.Я правильно думаю?и если да, то какому решению я должен следовать?спасибо 

def onMessageConsumed() = {
    stream.foreachRDD(rdd => {
      rdd.foreach(consumerRecord => {
        val record = new Record[T](consumerRecord.topic(), 
                                   consumerRecord.value())
        //notify observers with the record to compute
        notifyObservers(record)
      })
    })
  }

1 Ответ

0 голосов
/ 30 июня 2019

Да, классы, которые используются в коде, который отправляется другим исполнителям (выполняется в foreach и т. Д.), Должны реализовывать интерфейс Serializable.

также, если ваш код уведомления требует подключения к какому-либо ресурсу, вам нужно обернуть foreach в foreachPartition, что-то вроде этого: 

stream.foreachRDD(rdd => {
   rdd.foreachPartition(rddPartition =>
      // setup connection to external component      
      rddPartition.foreach(consumerRecord => {
        val record = new Record[T](consumerRecord.topic(), 
                                   consumerRecord.value())
        notifyObservers(record)
      })
      // close connection to external component
   })
  })
...