您可以通过使用KeyValueStore#approximateNumEntries()访问KTable的底层状态存储来获得每个分区中的近似计数，然后将此计数导出到prometheus（每个分区有一个计数）。
要访问底层状态存储，您可以使用低级处理器API通过每个StreamTask中的每个ProcessorContext（对应于一个分区）访问KeyValueStore。只需将KStream#transformValues()添加到您的拓扑中：

kStream
        ...
        .transformValues(ExtractCountTransformer::new, "your_ktable_name")
        ...

并在ExtractCountTransformer中提取计数给普罗米修斯：

@Log4j2
public class ExtractCountTransformer implements ValueTransformerWithKey<String, String, String> {

    private KeyValueStore<String, String> yourKTableKvStore;
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        yourKTableKvStore = (KeyValueStore<String, String>) context.getStateStore("your_ktable_name");
    }

    @Override
    public String transform(String readOnlyKey, String value) {
        //extract count to prometheus
        log.debug("partition {} - approx count {}", context.partition(), yourKTableKvStore.approximateNumEntries());
        yourKTableKvStore.approximateNumEntries();
        return value;
    }

    @Override
    public void close() {

    }
}

如果您公开了JMX指标，那么您可以获得许多Kafka指标，您正在寻找的指标是kafka_stream_state_estimate_num_keys。