Comment écrire à Kafka à partir de Spark Streaming

Oui, malheureusement, la Spark (1.x, 2.x) ne rend pas simple comment écrire à Kafka d'une manière efficace.

je te suggère la démarche suivante:

• Utiliser (et réutiliser)KafkaProducer instance par processus exécuteur / JVM.

Voici la configuration de haut niveau pour cette approche:

1. tout d'abord, vous devez "envelopper" Kafka's KafkaProducer parce que, comme vous l'avez mentionné, il n'est pas sérialisable. Emballage il vous permet de "l'envoyer" aux exécuteurs. L'idée clé ici est d'utiliser un lazy val de sorte que vous retardez l'instanciation du producteur jusqu'à sa première utilisation, ce qui est effectivement une solution pour que vous n'ayez pas à vous soucier de KafkaProducer n'étant pas sérialisable.
2. vous "expédiez" le producteur enveloppé à chaque exécuteur testamentaire en utilisant une variable de diffusion.
3. dans votre logique de traitement réel, vous accédez au producteur Enveloppé par la variable broadcast, et l'utilisez pour écrire les résultats du traitement retour à la Kafka.

les extraits de code ci-dessous fonctionnent avec Spark Streaming à partir de Spark 2.0.

Étape 1: Emballage KafkaProducer

import java.util.concurrent.Future

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}

class MySparkKafkaProducer[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {

  /* This is the key idea that allows us to work around running into
     NotSerializableExceptions. */
  lazy val producer = createProducer()

  def send(topic: String, key: K, value: V): Future[RecordMetadata] =
    producer.send(new ProducerRecord[K, V](topic, key, value))

  def send(topic: String, value: V): Future[RecordMetadata] =
    producer.send(new ProducerRecord[K, V](topic, value))

}

object MySparkKafkaProducer {

  import scala.collection.JavaConversions._

  def apply[K, V](config: Map[String, Object]): MySparkKafkaProducer[K, V] = {
    val createProducerFunc = () => {
      val producer = new KafkaProducer[K, V](config)

      sys.addShutdownHook {
        // Ensure that, on executor JVM shutdown, the Kafka producer sends
        // any buffered messages to Kafka before shutting down.
        producer.close()
      }

      producer
    }
    new MySparkKafkaProducer(createProducerFunc)
  }

  def apply[K, V](config: java.util.Properties): MySparkKafkaProducer[K, V] = apply(config.toMap)

}

Étape 2: Utilisez une variable de diffusion pour donner à chaque exécuteur son propre enveloppement KafkaProducer exemple

import org.apache.kafka.clients.producer.ProducerConfig

val ssc: StreamingContext = {
  val sparkConf = new SparkConf().setAppName("spark-streaming-kafka-example").setMaster("local[2]")
  new StreamingContext(sparkConf, Seconds(1))
}

ssc.checkpoint("checkpoint-directory")

val kafkaProducer: Broadcast[MySparkKafkaProducer[Array[Byte], String]] = {
  val kafkaProducerConfig = {
    val p = new Properties()
    p.setProperty("bootstrap.servers", "broker1:9092")
    p.setProperty("key.serializer", classOf[ByteArraySerializer].getName)
    p.setProperty("value.serializer", classOf[StringSerializer].getName)
    p
  }
  ssc.sparkContext.broadcast(MySparkKafkaProducer[Array[Byte], String](kafkaProducerConfig))
}

Etape 3: Ecrire à partir de Spark Streaming vers Kafka, en réutilisant le même wrapped KafkaProducer exemple (pour chaque exécuteur testamentaire)

import java.util.concurrent.Future
import org.apache.kafka.clients.producer.RecordMetadata

val stream: DStream[String] = ???
stream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val metadata: Stream[Future[RecordMetadata]] = partitionOfRecords.map { record =>
      kafkaProducer.value.send("my-output-topic", record)
    }.toStream
    metadata.foreach { metadata => metadata.get() }
  }
}

j'Espère cela aide.

Il y a un flux de Kafka, Écrivain maintenu par Cloudera (en fait le spin off à partir d'une Étincelle JIRA [1]