KSQL流处理中如何转换并保留复合键至下游流？

我明白你的痛点：当对源流的复合键字段做清洗转换后，下游流丢失了键信息，这会直接影响MySQL Sink Connector处理tombstone删除操作。下面是具体的解决方案，核心是显式指定下游流的键字段并通过PARTITION BY确保Kafka消息键被正确设置。

第一步：确保源流的键被KSQL正确识别 #

你的源流TEST_1来自数据库，带有复合键COL1和COL2，但默认情况下KSQL不会自动识别这些键字段。修改源流创建语句，显式声明键信息：

CREATE STREAM TEST_1 (COL1 STRING, COL2 STRING, COL3 STRING) 
WITH (
    KAFKA_TOPIC='TEST_1', 
    PARTITIONS=1, 
    REPLICAS=1, 
    VALUE_FORMAT='AVRO',
    KEY_FORMAT='AVRO', -- 匹配源topic的键格式（如源键是JSON则改为'JSON'）
    KEY_FIELDS='COL1,COL2' -- 明确标记复合键字段
);

第二步：创建转换后的下游流并保留键 #

在创建TEST_2时，需要完成两个关键操作来保留键：

1. 在WITH子句中声明新流的复合键为转换后的COL1和COL2
2. 使用PARTITION BY子句指定按转换后的键字段分区，这会让KSQL将这些字段写入Kafka消息的键中

修改后的创建语句：

CREATE STREAM TEST_2 
WITH (
    KAFKA_TOPIC='TEST_2', 
    PARTITIONS=1, 
    REPLICAS=1, 
    VALUE_FORMAT='AVRO',
    KEY_FORMAT='AVRO', -- 和源流保持一致，或按需调整
    KEY_FIELDS='COL1,COL2' -- 声明转换后的复合键字段
) AS SELECT 
    CLEAR(COL1) AS COL1, 
    CLEAR(COL2) AS COL2, 
    CLEAR(COL3) AS COL3 
FROM TEST_1
PARTITION BY COL1, COL2; -- 核心：确保转换后的字段被设为Kafka消息的键

为什么之前的实现丢失了键？ #

默认情况下，KSQL的CREATE STREAM ... AS SELECT不会自动将SELECT中的字段设为流的键——哪怕这些字段是源流的键。只有当你显式使用PARTITION BY指定字段，并且在WITH子句中用KEY_FIELDS声明键时，KSQL才会将这些字段作为Kafka消息的键写入下游topic。

确保Tombstone记录正常工作 #

为了支持MySQL的tombstone删除（即键存在但值为null的消息），需要确保你的自定义UDF CLEAR()能正确处理null值：

• 当输入为null时，CLEAR()应该返回null，这样tombstone记录的键会被正确转换，同时值保持null，MySQL Sink Connector就能识别为删除操作。

验证键是否正确保留 #

你可以用KSQL的PRINT命令验证TEST_2 topic中的消息键：

PRINT TEST_2 FROM BEGINNING LIMIT 10;

查看输出中的Key部分，应该能看到转换后的COL1和COL2的复合值，而不是默认的null或其他无效值。

内容的提问来源于stack exchange，提问作者Sanjay Nayak