Как использовать несколько счетчиков в Flink - PullRequest
Новая
       26

Как использовать несколько счетчиков в Flink

0 голосов
/ 18 октября 2019

(как-то связано с Как создать динамическую метрику в Flink )

У меня есть поток events(someid:String, name:String), и для целей мониторинга мне нужен счетчик на идентификатор события. Во всех документациях и примерах Flink я вижу, что счетчик, например, инициализируется именем в open функции карты.

Но в моем случае я не могу инициализировать счетчик какМне понадобится один для eventId, и я не знаю заранее значение. Кроме того, я понимаю, как дорого будет создавать новый счетчик каждый раз, когда в методе MapFunction используется map() метод даже. Наконец, я не могу сохранить «кэш» счетчиков, так как он будет слишком большим.

В идеале я хотел бы что-то вроде этого: 

class Event(id: String, name: String)

class ExampleMapFunction extends RichMapFunction[Event, Event] {
  @transient private var counter: Counter = _

  override def open(parameters: Configuration): Unit = {
    counter = new Counter()
  }

  override def map(event: Event): Event = {
    counter.inc(event.id)
    event
  }
}

Или в принципе я мог бы реализовать свой собственныйсчетчик, который позволяет мне пройти измерение? если да, то как?

Есть ли какие-либо советы или рекомендации для такого варианта использования?

1 Ответ

1 голос
/ 18 октября 2019

Если хранение кэша счетчиков будет слишком большим, то я не думаю, что использование метрик будет масштабироваться таким образом, чтобы удовлетворить ваши требования.

Несколько альтернатив:

  • Использование побочных выходов для сбора значимых событий в некотором внешнем, запрашиваемом / визуализируемом хранилище данных - например, effxdb.

  • Удерживайте информацию в нажатом состоянии и используйте широковещательные сообщения для запуска вывода соответствующих частей по желанию (опять же с использованием боковых выходов).

  • Удерживайте информацию всостояние ключа и периодически выбираются точки сохранения, которые затем анализируются с помощью запросов с использованием API процессора состояний.

...