Спарк кафка производитель сериализуемый

Я придумаю исключение:

ОШИБКА yarn.ApplicationMaster: Исключение класса пользователя: org.apache.spark.SparkException: Задача не сериализуема org.apache.spark.SparkException: Задача не сериализуема в org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304) в org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294) в org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122) в org.apache.spark.SparkContext.clean(SparkContext.scala:2032) в org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:889) в org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:888) в org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147) в org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108) в org.apache.spark.rdd.RDD.withScope(RDD.scala:306) в org.apache.spark.rdd.RDD.foreach(RDD.scala:888) в com.Boot$.test(Boot.scala:60) на com.Boot$.main(Boot.scala:36) на com.Boot.main(Boot.scala) at sun.reflect.NativeMethodAccessorImpl.invoke0(родной метод) в sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) в sun.reflect.DelegatingMethodAccessorImpl.intaho.jf.jj.invoke(Method.java:606) в org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:525) Причина: java.io.NotSerializableException: org.apache.kafka.clients.producer.KafkaProducer Стек сериализации: - объект не сериализуем (класс: org.apache.kafka.clients.producer.KafkaProducer, значение: org.apache.kafka.clients.producer.KafkaProducer@77624599) - поле (класс: com.Boot$$anonfun$test$1, имя: производитель $ 1, тип: класс org.apache.kafka.clients.producer.KafkaProducer) - объект (класс com.Boot$$anonfun$test$1,) в org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40) в org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47) at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:84) в org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:301)

//    @transient
val sparkConf = new SparkConf()

sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

//    @transient
val sc = new SparkContext(sparkConf)

val requestSet: RDD[String] = sc.textFile(s"hdfs:/user/bigdata/ADVERTISE-IMPRESSION-STAT*/*")

//    @transient
val props = new HashMap[String, Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, NearLineConfig.kafka_brokers)
//    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
//    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
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.put("producer.type", "async")
props.put(ProducerConfig.BATCH_SIZE_CONFIG, "49152")

//    @transient
val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](props)

requestSet.foreachPartition((partisions: Iterator[String]) => {
  partisions.foreach((line: String) => {
    try {
      producer.send(new ProducerRecord[String, String]("testtopic", line))
    } catch {
      case ex: Exception => {
        log.warn(ex.getMessage, ex)
      }
    }
  })
})

producer.close()

В этой программе я пытаюсь прочитать записи из пути hdfs и сохранить их в kafka. проблема в том, что когда я удаляю коды об отправке записей в кафку, все работает хорошо. Что я пропустил?

Источник

1 ответ

Решение

KafkaProducer не сериализуем. Вам нужно будет переместить создание экземпляра внутрь foreachPartition:

requestSet.foreachPartition((partitions: Iterator[String]) => {
  val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](props)
  partitions.foreach((line: String) => {
    try {
      producer.send(new ProducerRecord[String, String]("testtopic", line))
    } catch {
      case ex: Exception => {
        log.warn(ex.getMessage, ex)
      }
    }
  })
})

Обратите внимание, что KafkaProducer.send возвращает Future[RecordMetadata]и единственное исключение, которое может распространяться из него SerializationException если ключ или значение не могут быть сериализованы.

Источник

Я не рекомендую ответчика Юваля Ицчакова, потому что вы открываете и закрываете много сокетов, даже открытие соединения в брокере с kafka тяжелое и медленное, поэтому я настоятельно рекомендую прочитать этот блог https://allegro.tech/2015/08/spark-kafka-integration.html Я использовал его и протестировал, и это лучший вариант, который я также использовал в продуктивной среде.

Источник
Другие вопросы по тегам