#apache-kafka #apache-flink
#apache-kafka #apache-flink
Вопрос:
У меня есть задание Flink, которое считывает данные из Kafka (версия 0.9) и записывает в Redis. Я хочу отслеживать records-consumed-rate и records-lag-max метрики, передаваемые Kafka, которые Flink должен иметь возможность пересылать. В этом случае я пересылаю в Datadog.
Когда я запускаю задание с параллелизмом, равным 1, я вижу, что эта метрика генерируется просто отлично. Однако, если я сделаю параллелизм больше 1, эта метрика больше не пересылается. Задание выполняется при параллелизме> 1, потому что я вижу записи, записываемые в Redis.
Я запускаю Flink (v1.6.2) на AWS EMR:
- главный узел: (1) m4.large
- основной узел: (1) c4.2xlarge
- num.task.managers: 1
- slots.per.task.manager: 7
- параллелизм: 7
Параллелизм устанавливается StreamExecutionEnvironment.setParallelism() . Каждый потребитель Kafka создается с тем же group.id и уникальный client.id .
Агент DD отлично работает в кластере. Генерируется много метрик, таких как numberOfCompletedCheckpoints и время безотказной работы и т. Д.
Есть ли какая-либо причина, по которой Flink не будет пересылать эти метрики из Kafka, если параллелизм больше 1?
Обновление: я также попытался отправить пользовательскую метрику DD ( counter.inc() ) из функции Redis RichSinkFunction. Когда параллелизм = 1, метрика отправляется нормально. Когда параллелизм = 7, метрика не отправляется, однако она вызывается (добавлена строка отладки). Таким образом, кажется, что это не ограничивается пересылаемыми метриками из Kafka.
Комментарии:
1. Не уверен, как работает DataDog, поскольку я никогда им не пользовался, но когда вы устанавливаете параллелизм> 1, Flink отправляет отдельные метрики для каждого из TaskManagers, участвующих в обработке, так что, возможно, это проблема в вашем случае? Я использую Flink с Prometheus, и проблем с параллелизмом> 1 не было.
Ответ №1:
Проблема заключалась в том, что размер HttpRequest был слишком большим, чем выше параллелизм, что имеет смысл. Я возвращал «Объект запроса слишком большой», однако исключение не выходило из системы правильно, поэтому я пропустил его.
Похоже, что Flink DatadogHttpReporter не учитывает размер запроса при его создании. Я изменил Reporter, чтобы ограничить количество метрик на запрос 1000. Теперь метрики отображаются просто отлично.