Flink如何在未知所有key的情况下按名称删除指定状态全部条目

场景1：可接受作业重启（成本最低） #

分两种情况处理：

• 后续仍需使用该状态，仅需清空历史数据：直接修改ValueStateDescriptor的状态名称即可，Flink重启恢复时会按新名称创建空状态，旧状态数据会在后续savepoint中自动清理。
  示例修改：

  // 原定义
  // public static final ValueStateDescriptor<String> MY_STATE_DESCRIPTOR = 
  //        new ValueStateDescriptor<>("myState", String.class);
  // 修改为新的名称
  public static final ValueStateDescriptor<String> MY_STATE_DESCRIPTOR = 
          new ValueStateDescriptor<>("myState_v2", String.class);
  

  注意：如果使用了可查询状态，同步修改setQueryable的参数值即可
• 后续完全不需要使用该状态：直接删除所有和该状态相关的代码（包括状态描述符定义、状态变量声明、open方法中的初始化逻辑、业务逻辑中的读写代码），重启作业时添加--allow-non-restored-state启动参数，Flink会自动忽略无法匹配到对应描述符的旧状态，直接完成恢复。

场景2：作业不能重启，需运行时无感清空 #

可以通过Flink的KeyedStateBackend提供的applyToAllKeys方法遍历该状态下的所有key，逐个清理，实现步骤如下：

1. 在KeyedCoProcessFunction中声明KeyedStateBackend变量，open方法中完成初始化
2. 自定义触发逻辑（比如广播流控制信号、定时任务触发），收到清空指令时调用applyToAllKeys遍历所有key执行clear操作
   完整代码示例：

public static final ValueStateDescriptor<String> MY_STATE_DESCRIPTOR = 
        new ValueStateDescriptor<>("myState", String.class);

static {
    MY_STATE_DESCRIPTOR.setQueryable("QueryableMyState");
}

protected transient ValueState<String> myState;
// 新增KeyedStateBackend变量，泛型为你的作业实际key类型
private transient KeyedStateBackend<String> keyedStateBackend;

@Override
public void open(Configuration parameters) throws Exception {
    myState = getRuntimeContext().getState(MY_STATE_DESCRIPTOR);
    // 初始化KeyedStateBackend
    keyedStateBackend = (KeyedStateBackend<String>) getRuntimeContext().getKeyedStateBackend();
}

// 示例：在广播流处理方法中触发清空逻辑，你可以根据实际业务修改触发时机
@Override
public void processElement2(StreamRecord<ControlSignal> value, Context ctx, Collector<OutputType> out) throws Exception {
    ControlSignal signal = value.getValue();
    if (signal.isClearMyState()) {
        // 遍历该状态下所有key执行清空
        keyedStateBackend.applyToAllKeys(
                MY_STATE_DESCRIPTOR.getName(),
                MY_STATE_DESCRIPTOR.getTypeSerializer(),
                key -> {
                    myState.clear();
                    return null;
                }
        );
    }
}

注意：如果状态数据量较大，applyToAllKeys执行会消耗一定的IO和CPU资源，建议在业务低峰期触发，避免影响正常业务处理

注意事项 #

• 不建议直接手动删除状态后端（比如RocksDB、HDFS）上的状态文件，容易引发作业数据不一致甚至崩溃
• 生产环境执行状态清空操作前，建议先执行savepoint备份，避免误操作导致数据丢失

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