如何在Spark Streaming 2.3.1中将每条记录写入多个kafka主题？

Question

如何在Spark Streaming 2.3.1中把每条记录写到多个kafka主题？换句话说，我得到了5条记录和两个输出kafka主题，我希望这两个输出主题中都有5条记录。

这里的问题不谈论结构化流媒体案例。我正在寻找特定的结构化流媒体。

Answer 1

不确定您使用的是java还是scala。下面是为两个不同主题生成消息的代码。你得打电话

dataset.foreachPartition(partionsrows => {
      val props = new util.HashMap[String, Object]()
      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServer)
      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")
      val producer = new KafkaProducer[String, String](props)
      partionsrows.foreach(row => {
        val offerId = row.get(0).toString.replace("[", "").replace("]", "")
        val message1 = new ProducerRecord[String, String]("topic1", "message")
        producer.send(message1)
        val message2 = new ProducerRecord[String, String]("topic2",  "message")
        producer.send(message2)
      })
    })