如何识别并监控Flink中的State状态及获取其实时数值？

一、利用Flink内置Web Dashboard #

Flink自带的Web Dashboard（默认端口8081）是最便捷的监控入口：

• 找到目标Job后，进入Jobs页面的Operator详情，查看State的基础统计：包括状态占用的堆/堆外内存大小、状态条目数量，以及Checkpoint过程中状态的持久化情况。
• 针对Keyed State，还能在Task详情里查看每个Key Group的状态分布，快速定位大状态的来源。

二、自定义Metric追踪实时数值 #

如果需要监控具体State的实时数值（比如计数器、累加器的当前值），可以通过Flink的Metric系统自定义指标：

1. 继承RichFunction（如RichMapFunction），在open()方法中获取MetricGroup。
2. 根据需求注册对应类型的Metric：Gauge适合暴露实时数值，Counter用于计数场景。
3. 在业务逻辑中同步更新Metric的值，之后就能在Web Dashboard的Metrics页面查看，也可以对接Prometheus等监控系统做持久化和告警。

示例代码：

public class StateTrackingMap extends RichMapFunction<String, String> {
    private transient ValueState<Long> visitCountState;
    private transient Gauge<Long> currentVisitGauge;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 初始化State
        ValueStateDescriptor<Long> stateDesc = new ValueStateDescriptor<>("visitCount", Long.class);
        visitCountState = getRuntimeContext().getState(stateDesc);
        
        // 注册Gauge指标
        currentVisitGauge = getRuntimeContext().getMetricGroup().gauge("current-visit-count", () -> {
            try {
                return visitCountState.value() == null ? 0L : visitCountState.value();
            } catch (Exception e) {
                return 0L;
            }
        });
    }

    @Override
    public String map(String userId) throws Exception {
        Long count = visitCountState.value();
        count = count == null ? 1L : count + 1;
        visitCountState.update(count);
        return userId + " visited " + count + " times";
    }
}

三、通过Queryable State API实时查询 #

Flink提供的Queryable State功能支持在运行时直接查询指定Key的State值：

1. 先启用Queryable State：在flink-conf.yaml中设置queryable-state.enable: true，或者在代码中通过QueryableStateStream将目标State标记为可查询。
2. 编写客户端代码连接集群，指定Job ID、Operator名称、State名称和目标Key，获取实时State值。

示例客户端代码：

QueryableStateClient queryClient = new QueryableStateClient("flink-master", 9069);
CompletableFuture<ValueState<Long>> stateFuture = queryClient.getKvState(
    JobID.fromHexString("your-job-id"),
    "queryable-visit-count",
    "user_123",
    BasicTypeInfo.STRING_TYPE_INFO,
    new ValueStateDescriptor<>("visitCount", Long.class)
);

stateFuture.thenAccept(state -> {
    try {
        System.out.println("User 123's visit count: " + state.value());
    } catch (Exception e) {
        e.printStackTrace();
    }
});

四、分析Savepoint/Checkpoint中的State #

如果需要离线查看State的完整内容，可以触发Savepoint后用State Processor API解析：

• 命令行触发Savepoint：flink savepoint <job-id> <hdfs://path/to/savepoint>
• 编写代码加载Savepoint，读取并分析其中的State数据，适合排查历史状态问题。

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