flink sink多个topic

互联网 12 月前 0 2

本文介绍: 在上面的代码中，我们首先定义了一个数据流 stream，其中元素类型为，表示 Kaf k a Top ic 和消息内容。接着，我们定义了一个 Kaf k a 序列化器，用于将数据流中的元素转换成 Kaf k a Pro du ce rRe co r d。最后，我们创建了一个 FlinkKaf k aPro du ce r 实例，并将数据流 stream 发送到 Kaf k a 中。

flink str e a m 数据 动态 写入 多个 topic

flink1.15之前

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaProducer, KafkaSerializationSchema}
import org.apache.kafka.clients.producer.ProducerRecord

object DynamicKafkaProducer {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 数据流中的元素类型为 (topic: String, message: String)
    val stream: DataStream[(String, String)] = ...

    // 定义 Kafka 序列化器
    val kafkaSerializationSchema = new KafkaSerializationSchema[(String, String)] {
      override def serialize(element: (String, String), timestamp: java.lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
        new ProducerRecord(element._1, element._2.getBytes("UTF-8"))
      }
    }

    // 创建 FlinkKafkaProducer 实例
    val kafkaProducer = new FlinkKafkaProducer[(String, String)](
      "localh