如何在Flink流作业中通过KafkaSink实现同Key数据写入Kafka同一分区

要实现相同uniqueField值的数据进入Kafka同一分区，核心是通过自定义KafkaRecordSerializationSchema，从JSON字符串中提取uniqueField作为Kafka消息的Key——Kafka默认分区器会根据Key的哈希值分配分区，从而保证同Key的消息进入同一分区。

步骤1：自定义序列化Schema #

使用Jackson解析JSON字符串（Flink项目通常已依赖Jackson），实现KafkaRecordSerializationSchema接口，将uniqueField作为消息Key：

import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema
import org.apache.flink.connector.kafka.sink.KafkaSinkContext
import org.apache.kafka.clients.producer.ProducerRecord
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.DeserializationFeature
import java.nio.charset.StandardCharsets

class KeyedKafkaSerializationSchema(topic: String) extends KafkaRecordSerializationSchema[String] {
  // 初始化Jackson ObjectMapper，忽略未知字段避免解析异常
  private val objectMapper = new ObjectMapper()
  objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)

  override def serialize(
    element: String,
    context: KafkaSinkContext,
    timestamp: java.lang.Long
  ): ProducerRecord[Array[Byte], Array[Byte]] = {
    // 解析JSON字符串，提取uniqueField
    val jsonNode = objectMapper.readTree(element)
    val uniqueKey = jsonNode.get("uniqueField").asText() // 若uniqueField是数字类型，可改用asLongValue()

    // 创建ProducerRecord，设置Key、Value、Topic和时间戳
    new ProducerRecord[Array[Byte], Array[Byte]](
      topic,
      null, // 分区设为null，由Kafka默认分区器根据Key分配
      timestamp,
      uniqueKey.getBytes(StandardCharsets.UTF_8),
      element.getBytes(StandardCharsets.UTF_8)
    )
  }
}

步骤2：修改KafkaSink构建逻辑 #

替换原代码中默认的序列化Schema为自定义实现：

import org.apache.flink.connector.kafka.sink.KafkaSink
import org.apache.flink.connector.kafka.sink.DeliveryGuarantee
import java.util.Properties

def getFlinkKafkaSink(jobConf: JobConfig, kafkaInfo: KafkaInfo): KafkaSink[String] = {
  val properties: Properties = getProperties()
  KafkaSink
    .builder[String]
    .setBootstrapServers(kafkaInfo.bootstrapServers)
    // 使用自定义序列化Schema
    .setRecordSerializer(new KeyedKafkaSerializationSchema(kafkaInfo.topics))
    .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
    .setKafkaProducerConfig(properties)
    .build()
}

关键说明 #

• 异常处理：若存在JSON格式错误或缺失uniqueField的情况，可在serialize方法中添加try-catch块，根据业务需求选择抛出异常、使用默认Key或丢弃消息。
• Key类型适配：如果uniqueField是数字（如Long），可将asText()改为asLongValue()，再转成字节数组（如String.valueOf(uniqueKey).getBytes(...)），确保相同数值生成相同Key字节数组。
• 分区器逻辑：Kafka默认分区器会将相同Key的消息路由到同一分区，若需自定义分区规则，可在ProducerRecord中显式指定partition参数，或配置Kafka生产者的partitioner.class属性。

内容的提问来源于stack exchange，提问作者Alok Singh