基于内容的Flink SQL去重：替代无界窗口的方案咨询

你的核心需求是仅让id和segments为新值或与前值不同的记录流向下游，现有方案用无界LAG算子导致状态持续膨胀，以下是几种更优的替代实现方式：

方案1：给无界LAG算子配置状态TTL #

既然现有逻辑能正常运行，最简单的优化是给算子设置状态过期时间，避免状态无限增长。Flink支持通过SQL hint或全局配置为状态设置TTL。

实现方式 #

修改table_one_source_keyed视图，添加状态TTL的hint：

Create or replace view table_one_source_keyed as (
    WITH table_one_source_hash AS (
        SELECT
            pod, org, tenantId, pk, id, segments, ts,
            HASH_CODE(tenantId || id || CAST(segments AS STRING)) AS data_hash
        FROM table_one_source
    ),
    entitlement_source_deduped as (
         SELECT *
         FROM (
             SELECT *,
                    -- 添加状态TTL hint，设置过期时间（示例为7天）
                    /*+ OPTIONS('state.ttl'='7 d') */
                    LAG(data_hash) OVER (PARTITION BY tenantId, id ORDER BY ts) AS prev_data_hash
             FROM table_one_source_hash
             )
         WHERE data_hash IS DISTINCT FROM prev_data_hash OR prev_data_hash IS NULL
    )
    
select * from entitlement_source_deduped
);

也可以通过全局配置统一设置：

SET table.exec.state.ttl = '7 d';

优缺点 #

• 优点：改动极小，无需调整核心逻辑，快速解决状态膨胀问题
• 缺点：若业务需要永久保留所有历史对比记录，TTL不适用；需根据业务场景合理设置过期时间

方案2：使用基于时间窗口的DISTINCT去重 #

如果业务允许仅在固定时间窗口内去重（比如最近1小时内的重复记录过滤），可以用DISTINCT结合滑动/滚动窗口实现，状态仅保留窗口内的数据。

实现方式 #

Create or replace view table_one_source_keyed as (
    SELECT 
        pod, org, tenantId, pk, id, segments, ts
    FROM table_one_source
    -- 按tenantId+id分区，按ts做1小时滚动窗口
    WINDOW TUMBLING (PARTITION BY tenantId, id ORDER BY ts INTERVAL '1' HOUR)
    -- 基于tenantId、id、segments的组合去重
    DISTINCT tenantId, id, segments
);

优缺点 #

• 优点：状态大小受窗口大小限制，不会无限膨胀；SQL逻辑简洁
• 缺点：仅能过滤窗口内的重复记录，若跨窗口出现重复（比如1小时前的相同记录）会被放行，不符合全量去重需求时不适用

方案3：使用Keyed State UDF手动管理对比状态 #

通过自定义UDF实现更精细的状态控制，仅保留每个tenantId+id的最新segments哈希值，对比新记录后更新状态，无需保留所有历史数据。

实现步骤 #

1. 定义一个Scalar UDF，维护每个key的最新哈希值：

public class LatestHashChecker extends ScalarFunction {
    private ValueState<Integer> latestHashState;

    @Override
    public void open(FunctionContext context) throws Exception {
        ValueStateDescriptor<Integer> desc = new ValueStateDescriptor<>("latestHash", Integer.class);
        // 可选：设置状态TTL
        desc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(7)).build());
        latestHashState = getRuntimeContext().getState(desc);
    }

    public boolean checkIfChanged(int currentHash) throws Exception {
        Integer prevHash = latestHashState.value();
        if (prevHash == null || prevHash != currentHash) {
            latestHashState.update(currentHash);
            return true;
        }
        return false;
    }
}

2. 在Flink SQL中注册UDF并使用：

CREATE FUNCTION check_hash_change AS 'com.your.package.LatestHashChecker';

Create or replace view table_one_source_keyed as (
    SELECT 
        pod, org, tenantId, pk, id, segments, ts
    FROM (
        SELECT
            *,
            HASH_CODE(tenantId || id || CAST(segments AS STRING)) AS data_hash,
            -- 按tenantId+id分区，调用UDF判断是否变化
            check_hash_change(data_hash) OVER (PARTITION BY tenantId, id) AS is_changed
        FROM table_one_source
    )
    WHERE is_changed = true
);

优缺点 #

• 优点：状态仅保留每个key的最新哈希值，内存占用极小；可自定义状态TTL和清理逻辑
• 缺点：需要编写Java/Scala UDF，相比纯SQL实现复杂度更高

方案4：利用上游Kafka的幂等性/Exactly-Once语义 #

如果上游生产者能保证相同tenantId+id的重复记录不会重复写入Kafka（比如开启Kafka生产者幂等性或事务），则可以直接跳过Flink端的去重逻辑，减少状态开销。

实现方式 #

无需修改现有SQL，只需确保上游生产者配置：

# Kafka生产者核心配置
enable.idempotence=true
transactional.id=your-unique-transaction-id

优缺点 #

• 优点：Flink端无需任何去重逻辑，性能最优
• 缺点：依赖上游系统的可靠性，若上游无法保证幂等性则不适用

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