java - 如何在 Flink 中增加 SinkFunction 的 numRecordsOutPut 指标？

Question

我正在使用 flink 消费 kafka 并写入 redis。

这是我的 redis 接收函数:

            .addSink(new RichSinkFunction<MobilePageEvent>() {

                @Override
                public void invoke(MobilePageEvent event, Context context) {

                    JEDIS_CLUSTER.zadd(..);
                }
            })
            .name("redis sink");

虽然我可以从 redis 命令行获取数据，但指标显示 sink 函数的输出为零:

我怎样才能增加这个指标？

Answer 1

numRecordsIn 和 numRecordsOut 指标仅计算在 Flink 作业本身内流动的流记录，不包括与外部系统的通信。所以换句话说，sources 不报告任何传入的记录，而 sinks 不报告任何传出的记录。

在我看来，您有几个选择:

1. 使用接收器上的 numRecordsIn 指标作为你想知道的近似值
2. 派生或扩展 RedisSink 并添加您想要的指标

显示了添加计数器指标的模式 here .

对于 redis 接收器，您可以在 open() 方法中初始化一个 Counter，然后在 invoke() 中递增它。但这似乎毫无意义，因为这只会反射(reflect) numRecordsIn 指标。如果您的 Redis 接收器正在执行缓冲的批量写入，那么等待递增指标直到数据实际发送到 Redis 可能更有意义——在这种情况下，您可能宁愿使用 Meter 而不是 Counter。