Чтобы сделать наши курсы по Apache Flink для дата-инженеров и разработчиков распределенных приложений еще более полезными, сегодня рассмотрим, как этот фреймворк потоковой аналитики больших данных реализует концепцию оконных функций. Жизненный цикл окна, ключевые понятия и оконные операции Apache Flink, управляемые данными и временем.
Что такое окно в потоковой обработке данных вообще и в Apache Flink в частности
Будучи распределенным фреймворком для создания потоковых stateful-приложений, Apache Flink активно использует концепцию окон, которая лежит в основе обработки бесконечных потоков данных. Окно разбивает поток на «сегменты» конечного размера, над которыми выполняются вычисления. Окно создается, как только появляется первый элемент, который должен принадлежать ему, и полностью удаляется, когда время события или обработки проходит отметку окончания (timestamp) плюс заданное пользователем допустимое время задержки (Allowed Lateness). Flink гарантирует удаление только для временных окон, но не для других типов, таких как глобальные окна. Например, при использовании оконной стратегии, основанной на времени события, которая создает неперекрывающиеся (или переворачивающиеся) окна каждые 5 минут и имеет допустимую задержку в 1 минуту, Flink создаст новое окно для интервала между 12:00 и 12:00: 05, когда прибудет первый элемент с временной меткой, попадающей в этот интервал. И удалит его, когда водяной знак пройдет временную метку 12:06.
Кроме того, каждое окно будет иметь триггер и функцию: ProcessWindowFunction, ReduceFunction или AggregateFunction. Функция будет содержать вычисление, которое будет применено к содержимому окна, а триггер задает условия, при которых окно считается готовым для применения функции. Политика срабатывания может быть чем-то вроде «когда количество элементов в окне больше 4» или «когда водяной знак пересекает конец окна». Триггер также может решить очистить содержимое окна в любой момент между его созданием и удалением. Очистка в этом случае относится только к элементам в окне, а не к метаданным окна. Это означает, что новые данные все еще могут быть добавлены в это окно. Также можно указать Evictor, который будет удалять элементы из окна после срабатывания триггера, а также до и/или после применения функции.
Перед определением окна следует указать наличие ключа для него. Использование keyBy(…) разделит бесконечный поток на потоки с логическими ключами. В качестве ключа может использоваться любой атрибут входящих событий. Наличие потока с ключом позволит оконным вычислениям выполняться параллельно несколькими задачами, поскольку каждый логический поток с ключом может обрабатываться независимо от остальных. Все элементы, относящиеся к одному и тому же ключу, будут отправлены в одну и ту же параллельную задачу. Без использования ключа исходный поток не будет разделен на несколько логических потоков, и вся оконная логика будет выполняться одной задачей, то есть с параллелизмом 1.
Далее следует определить средства назначения окна, указав WindowAssigner в вызове window(…) для потоков с ключом или windowAll() в противном случае. WindowAssigner отвечает за назначение каждого входящего элемента одному или нескольким окнам. Flink поставляется с предопределенными назначателями окон для наиболее распространенных случаев использования, а именно с переворачивающимися окнами, скользящими окнами, окнами сеансов и глобальными окнами. Также разработчик может реализовать собственное средство назначения окон, расширив класс WindowAssigner. Все встроенные средства назначения окон, кроме глобальных, назначают элементы окнам на основе времени, которое может быть временем обработки или временем события.
Временные окна имеют отметку времени начала (включительно) и окончания (исключительно), которые вместе описывают размер окна. В коде Flink использует TimeWindow при работе с временными окнами, которые имеют методы для запроса метки времени начала и окончания, а также дополнительный метод maxTimestamp(), который возвращает самую большую разрешенную метку времени для данного окна. Далее рассмотрим основные оконные операции, управляемые данными и временем.
Основные оконные операции
Окно, управляемое данными запускается, как только оно достигнет точки, где требуемое количество данных получено из потока. Во Flink этого можно добиться с помощью функций countWindow() и countWindowAll(). Например, запуск функции суммирования каждый раз, когда количество элементов, введенных в сокет, достигает 3. Это реализуется с помощью функции countWindowAll(3). Входной поток может быть любым, однако в рассматриваемом примере сокет TextStream создан с использованием nc -lk 9999. Пример на Flink Scala, запускаемый в shell-оболочке, выглядит следующим образом:
val dataStream = senv.socketTextStream("localhost",9999,' ')
val outputSink = dataStream.flatMap(_.split(" "))
.map(_.toInt).countWindowAll(3)
outputSink.print()
senv.execute("Program to implement countWindowAll")
Как уже было отмечено выше, countWindowAll является окном без ключа, поскольку окна не группируются вместе с использованием операции keyBy(). А countWindow называется окнами с ключами, где элементы сгруппированы на основе заданного условия. Ниже показан пример на Flink Scala для countWindow, где элементы сгруппированы с использованием условия %2. В отличие от countWindowAll, окно будет запускаться всякий раз, когда счетчик соответствующего окна достигает 3, число будет суммировано и отображено.
val dataStream = senv.socketTextStream("localhost",9999,' ')
val keyedDataStream = dataStream.flatMap(_.split(" "))
.map(_.toInt).keyBy(a => a%2)
val outputSink = keyedDataStream.countWindow(3).sum(0)
outputSink.print()
senv.execute("Program to implement countWindow through Even/Odd No")
Стоит отметить, что Flink поддерживает различные типы временных окон:
- Переворачивающееся (Tumbling Time Window), которое определяет окно заданной продолжительности времени. Элементы сгруппированы по времени их поступления, и каждый элемент принадлежит ровно одному окну. Этот тип окна не перекрывается, т. е. события в одном окне не будут перекрываться в других окнах. Следующий код на Scala создает переворачивающееся окно в 5 секунд. В течение 5 секунд, благодаря операции keyBy(), нечетные количества событий помещаются в отдельные окна и суммируются.
val dataStream = senv.socketTextStream("localhost",9999,' ')
val keyedDataStream = dataStream.flatMap(_.split(" ")).map(_.toInt)
.keyBy(a => a%2)//Creates tumbling window of 5 seconds
val tumblingWindow = keyedDataStream.timeWindow(Time.seconds(5))
val outputSink = tumblingWindow.sum(0)
outputSink.print()
senv.execute("Program to implement tumbling window")
- Скользящее временное окно (Sliding Time Window) — в отличие от переворачивающегося временного окна, скользящее окно скользит по входящему потоку данных и может быть перекрывающимся. Пример кода на Scala для скользящего окна в 20 секунд с 10-секундным периодом скольжения выглядит так:
val dataStream = senv.socketTextStream("localhost",9999,' ')
val keyedDataStream = dataStream.flatMap(_.split(" "))
.map(_.toInt).keyBy(a => a%2)//Creates sliding window of 20 seconds with 10 seconds slide
val slidingWindow = keyedDataStream.timeWindow(Time.seconds(20)
,Time.seconds(10))
val outputSink = slidingWindow.sum(0)
outputSink.print()
senv.execute("Program to implement sliding window")
В следующий раз мы продолжим разговор про конные операции и метки времени в Apache Flink и разберемся с понятием водяного знака. Как упростить работу с заданиями Flink SQL с помощью проекта SeaTunnel, читайте в нашей новой статье. А освоить практику использования возможностей Apache Flink для потоковой обработки событий в распределенных приложениях аналитики больших данных вам помогут специализированные курсы в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
