Как увеличить numRecordsOutPut метрики SinkFunction в Flink?

Question

Я использую flink для использования kafka и записи в redis.

Вот моя функция приемника для redis:

            .addSink(new RichSinkFunction<MobilePageEvent>() {

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

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

Хотя я могу получить данные из командной строки redis,но метрика показывает, что выход функции приемника равен нулю:

Как я могу увеличить эту метрику?

Answer 1

Метрики numRecordsIn и numRecordsOut учитывают только записи потока, которые передаются в самом задании Flink, и не включают связь с внешними системами.Другими словами, источники не сообщают о поступлении каких-либо записей, а приемники не сообщают о выходе каких-либо записей.

На мой взгляд, у вас есть несколько вариантов:

1. использовать метрику numRecordsIn в приемнике как приближение к тому, что вы хотите знать
2. fork илирасширить RedisSink и добавить нужную метрику

Шаблон для добавления метрики счетчика показан здесь .

В случае приемника redis вы можете инициализировать Counter в методе open () и увеличивать его в invoke ().Но это кажется довольно бессмысленным, поскольку это просто зеркально отражает показатель numRecordsIn.Если ваш приемник redis выполняет буферизованные массовые записи, то может быть более целесообразно подождать, чтобы увеличить метрику, пока данные не будут фактически отправлены на redis - и в этом случае вы могли бы скорее использовать Meter, а не Counter.