flink作业使用KafkaSink序列化消息key

一：定义Kafka序列化器

    public static class KafkaKeySerializer implements Serializer<Tuple2<String, String>>, Configurable {
        private String encoding = "UTF8";

        @Override
        public void configure(Map<String, ?> configs, boolean isKey) {
            Object encodingValue = configs.get("serializer.encoding");
            if (encodingValue instanceof String)
                encoding = (String) encodingValue;
        }

        @Override
        public byte[] serialize(String topic, Tuple2<String, String> data) {
            try {
                if (data == null)
                    return null;
                else
                    return data.f0.getBytes(encoding);
            } catch (UnsupportedEncodingException e) {
                throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding);
            }
        }

        @Override
        public void configure(Map<String, ?> configs) {
            Object encodingValue = configs.get("serializer.encoding");
            if (encodingValue instanceof String)
                encoding = (String) encodingValue;
        }
    }
    

二：模拟Flink作业运行逻辑

    public static void main(String[] args) throws Exception {
        // 1. 创建 Flink 执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 构造包含 key 和 value 的 DataStream 示例（key：客户ID，value: 消息内容）
        DataStream<Tuple2<String, String>> source = env.fromElements(
                Tuple2.of("customer1", "Hello from customer1"),
                Tuple2.of("customer2", "Hello from customer2"),
                Tuple2.of("customer1", "Another msg for customer1")
        );
        // 3. 自定义 key 序列化 Schema
        //    此处将 Tuple2 的第一个元素（客户ID）作为 key，通过 UTF-8 编码转换成字节数组
        SerializationSchema<Tuple2<String, String>> keySerializationSchema = element ->
                element.f0.getBytes(StandardCharsets.UTF_8);
        // 4. 自定义 value 序列化 Schema
        //    将 Tuple2 的第二个元素（消息内容）转换为字节数组
        SerializationSchema<Tuple2<String, String>> valueSerializationSchema = element ->
                element.f1.getBytes(StandardCharsets.UTF_8);
        // 5. 安装 KafkaRecordSerializationSchema 构建器，并配置 topic、value 序列化以及 key 的序列化方式
        KafkaRecordSerializationSchema<Tuple2<String, String>> recordSchema = KafkaRecordSerializationSchema
                .<Tuple2<String, String>>builder()
                // 指定写入的 topic
                .setTopic("test-topic")
                // 设置 value 的序列化器
                .setValueSerializationSchema(valueSerializationSchema)
                // 方法1: 通过指定 Serializer 类以及配置信息来设置 key 序列化器
                .setKafkaKeySerializer(KafkaKeySerializer.class, new HashMap<String, String>() {{
                    put("serializer.encoding", "UTF8");
                }})
                // 方法2: 直接通过指定 Serializer 类来设置 key 序列化器
                .setKafkaKeySerializer(KafkaKeySerializer.class)
                // 方法3: 设置 key 的 SerializationSchema，用于从记录中提取 key 并进行序列化
                .setKeySerializationSchema(keySerializationSchema)
                .build();
        // 6. 构建 KafkaSink，并指定 Kafka broker 地址与前面配置好的 Record Serialization Schema
        KafkaSink<Tuple2<String, String>> kafkaSink = KafkaSink
                .<Tuple2<String, String>>builder()
                .setBootstrapServers("localhost:9092")
                .setRecordSerializer(recordSchema)
                .build();
        // 7. 将数据流写入 Kafka
        source.sinkTo(kafkaSink);
        // 8. 执行 Flink 任务
        env.execute("Flink KafkaSink Key Example");
    }

提示：文中示例数据泛型为Tuple2<String, String>，如果数据为STRING，需要按照上述逻辑重新调整。