如何通过Flink SQL删除状态存储中指定tenant_id的记录？

针对你的Flink SQL左连接作业，要实现从Kafka收到删除事件后清理对应tenant_id的所有状态，可参考以下几种方案：

方案一：状态TTL结合删除事件标记 #

利用Flink状态TTL的自动清理机制，结合删除事件标记目标租户数据，触发快速清理：

• 1. 为table_1和table_2新增is_deleted字段（默认值false），用于标记状态是否需要清理。
• 2. 从删除事件的Kafka主题创建动态表delete_events，结构包含tenant_id字段。
• 3. 修改原有查询逻辑，关联delete_events将匹配的租户记录标记为已删除：

  INSERT INTO sink_table 
      SELECT r.field1,
             r.tenant_id,
             r.field2,
             r.field3,
             d.field4
      from (
          SELECT t1.*, CASE WHEN de.tenant_id IS NOT NULL THEN true ELSE false END AS is_deleted
          FROM table_1 t1
          LEFT JOIN delete_events de ON t1.tenant_id = de.tenant_id
      ) r
               LEFT JOIN (
          SELECT t2.*, CASE WHEN de.tenant_id IS NOT NULL THEN true ELSE false END AS is_deleted
          FROM table_2 t2
          LEFT JOIN delete_events de ON t2.tenant_id = de.tenant_id
      ) d
                         ON r.tenant_id = d.tenant_id AND r.field1 = d.field1
  

• 4. 通过Table API为两张表配置状态TTL，当is_deleted为true时触发立即过期：

  // 以table_1为例配置状态TTL
  Table table1 = tableEnv.from("table_1");
  tableEnv.getConfig().getConfiguration().set(
      "table.exec.state.ttl", "1s" // 设置极短TTL，标记删除后立即清理
  );
  

• 5. 调整Flink参数state.cleanup.interval，确保过期状态能被及时清理。

方案二：使用Flink SQL流处理DELETE语句（需CDC源支持） #

如果你的Flink版本≥1.12，且table_1、table_2为CDC格式源（如Debezium），可直接通过DELETE语句触发状态清理：

• 1. 从删除事件Kafka主题创建临时表delete_tenants：

  CREATE TABLE delete_tenants (
      tenant_id STRING,
      event_time TIMESTAMP(3) METADATA FROM 'timestamp'
  ) WITH (
      'connector' = 'kafka',
      'topic' = 'your-delete-topic',
      'properties.bootstrap.servers' = 'your-kafka-server',
      'format' = 'json'
  );
  

• 2. 执行DELETE语句删除对应租户的记录，Flink会自动清理关联的状态：

  DELETE FROM table_1 WHERE tenant_id IN (SELECT tenant_id FROM delete_tenants);
  DELETE FROM table_2 WHERE tenant_id IN (SELECT tenant_id FROM delete_tenants);
  

• 注意：必须开启Checkpoint，状态清理会在Checkpoint完成后生效。

方案三：自定义ProcessFunction手动管理状态（灵活性最高） #

若SQL层面方案无法满足需求，可转为Table API+DataStream混合模式，手动处理状态清理：

• 1. 将table_1和table_2转换为DataStream，同时将删除事件的Kafka主题作为侧输入（Side Input）。
• 2. 自定义ProcessFunction，维护以(tenant_id, field1)为键的状态，收到删除事件时遍历并清理目标租户的所有状态：

  public class TenantStateCleanupProcess extends KeyedProcessFunction<Tuple2<String, String>, YourRecord, YourOutput> {
      private MapState<String, YourState> state;
  
      @Override
      public void open(Configuration parameters) {
          MapStateDescriptor<String, YourState> stateDesc = new MapStateDescriptor<>(
              "tenant-state",
              BasicTypeInfo.STRING_TYPE_INFO,
              TypeInformation.of(YourState.class)
          );
          state = getRuntimeContext().getMapState(stateDesc);
      }
  
      @Override
      public void processElement(YourRecord value, Context ctx, Collector<YourOutput> out) throws Exception {
          // 处理业务逻辑并更新状态
          state.put(value.getField1(), new YourState(value));
          out.collect(...);
      }
  
      @Override
      public void processBroadcastElement(DeleteEvent deleteEvent, Context ctx, Collector<YourOutput> out) throws Exception {
          // 遍历并删除目标tenant_id下的所有状态
          Iterator<Map.Entry<String, YourState>> iterator = state.iterator();
          while (iterator.hasNext()) {
              Map.Entry<String, YourState> entry = iterator.next();
              if (entry.getValue().getTenantId().equals(deleteEvent.getTenantId())) {
                  iterator.remove();
              }
          }
      }
  }
  

• 3. 将主数据流与侧输入连接，应用自定义ProcessFunction。

核心注意事项 #

• 所有方案必须开启Checkpoint，状态变更仅在Checkpoint完成后才会持久化。
• 你的JOIN作业实际基于tenant_id + field1复合键存储状态，清理时需覆盖该租户下所有field1对应的状态。
• 使用状态TTL时，需设置table.exec.state.ttl.ignore-keys为false，保证键级别的TTL生效。

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