如何使用KSQL避免流中重复输出相同事件？

方法1：用DISTINCT结合状态存储实现自动去重 #

KSQL内置的DISTINCT关键字会基于状态存储跟踪已输出的记录组合，直接过滤重复的(A,B,C)事件：

CREATE STREAM deduplicated_stream AS
SELECT DISTINCT A, B, C
FROM input_stream
PARTITION BY A
EMIT CHANGES;

这里PARTITION BY A确保同一键的事件路由到同一任务，状态存储会独立维护每个键下已输出的(B,C)组合，后续收到相同组合的事件会直接被过滤。

方法2：自定义状态过期的去重（适合非永久去重场景） #

如果需要自动清理过期状态，避免存储无限膨胀，可结合窗口和GROUP BY实现：

CREATE STREAM deduplicated_stream_with_ttl AS
SELECT A, B, C
FROM input_stream
PARTITION BY A
GROUP BY A, B, C
HAVING COUNT(*) = 1  -- 仅保留首次出现的组合
EMIT CHANGES
WITH (
  KAFKA_TOPIC='deduplicated_topic',
  VALUE_FORMAT='JSON',
  WINDOW_TYPE='SESSION',
  WINDOW_SIZE='1 DAYS'  -- 1天后自动清理该键下的过期状态
);

方法3：基于哈希对比的去重 #

通过计算(A,B,C)组合的哈希值，快速判断是否重复，减少状态存储的内容：

CREATE STREAM deduplicated_stream_hash AS
SELECT A, B, C, HASH(A, B, C) AS record_hash
FROM input_stream
PARTITION BY A
GROUP BY A, B, C, HASH(A, B, C)
HAVING COUNT(*) = 1
EMIT CHANGES;

哈希值会作为快速对比的标识，状态存储只需跟踪哈希码+键，降低存储开销。

最佳实践 #

• 优先用内置DISTINCT：KSQL底层已基于Kafka Streams优化状态管理，代码简洁且性能稳定，无需手动维护哈希或状态逻辑。
• 强制设置状态TTL：流处理的状态会持续积累，必须通过窗口或STATE_STORE_RETENTION_MS参数设置过期时间（比如86400000毫秒=1天），避免资源耗尽。
• 按业务键分区：必须用PARTITION BY A（原始事件的键），确保相同键的事件集中处理，状态不分散，去重逻辑准确。
• 避免全局去重：不要尝试跨所有分区做全局去重，这会导致状态集中到单个任务，引发性能瓶颈，尽量按业务键分区实现局部去重。
• 测试边界场景：验证同一键下连续重复事件、间隔多天的重复事件的去重效果，确保状态清理和去重逻辑正常工作。

内容的提问来源于stack exchange，提问作者Erik Schumann