В этой статье для обучения дата-инженеров и разработчиков приложений потоковой аналитики больших данных рассмотрим, на что следует обратить внимание при развертывании Apache Flink в реальных проектах. Обработка опоздавших данных, тонкости сериализации, проблемы неравномерного распределения и большие состояния заданий.
Обработка опоздавших данных в Apache Flink
В потоковой обработке данных, которую поддерживает Apache Flink, важно понимать не только время прихода события в систему, но и момент его происхождения в реальном мире. В идеально случае считается, что неважно, когда событие пришло в систему, поскольку его всегда можно связать с нужным временным окном, даже при выходе приложения из строя. Но с технической точки зрения это не так просто. Например, триггер временного окна по умолчанию будет запускать окно для каждого события, пришедшего с опозданием. Если в конвейере, который обрабатывает миллионы событий в секунду, один из продюсеров «завис» на час, то система будет перегружена устаревшими данными. А обработка событий, пришедших в систему с опозданием, на практике обходится довольно дорого.
Можно решить эту проблему путем реализации пользовательского триггера окна, что потребует разработки собственного кода на Java/Scala. Это сложнее SQL-запросов и требует знания особенностей Apache Flink. Кроме того, еще одна распространенная проблема поздних данных — это не заданное время жизни (Time To Live, TTL) для состояния окна. Если допускается опоздание на 3 дня, состояние для всех временных окон, созданных в течение этого периода времени, будет сохранено, поскольку стратегия обработки поздних данных в Apache Flink по умолчанию заключается в объединении окон, и для этого необходимо состояние задания.
При работе со временем также следует помнить о незанятых разделах и последствиях неупорядоченных интервалов. Бездействующие разделы тесно связаны с поздней обработкой данных и, чтобы работать с ними, нужно учитывать их водяной знак, что мы рассматривали здесь. Это снова может привести к поздним данным в случае, если события задержались на стороне вышестоящего приложения-продюсера.
Тонкости сериализации
В любой распределенной среде данные передаются по сети, при этом на каждом узле выполняется их сериализация – перевод в байт-код. При сбое сериализации POJO или Apache AVRO фреймворк Flink использует Kryo, но это может в десятки раз снижать эффективность. Поэтому рекомендуется отключить конфигурацию pipe.auto-type-registration, о чем мы рассказываем здесь. Кроме того, из-за накладных расходов на сериализацию можно столкнуться со снижением производительности при внесении изменений, направленных на ее оптимизацию. Желательно выполняться shuffle-операции, т.е. перетасовку данных прямо перед приемником, чтобы уменьшить количество подключений к внешней системе.
Например, если используется партиционирование для дампов данных во внешнюю файловую систему, каждая подзадача StreamingFileSink поддерживает список разделов вместе с файлом, куда записываются события. При случайном перемешивании данных это может привести к ситуациям, когда каждая подзадача записывает данные в каждый раздел. Чтобы избежать этого, лучше перемешивать данные на стороне Flink. Но при этом Flink будет сериализовать и десериализовать каждое событие, даже если обмен будет происходить внутри одной и той же JVM. Поэтому важно отслеживать системные метрики, о чем мы писали в этой статье.
Проблемы неравномерного распределения данных
Apache Flink обеспечивает четкую визуализацию количества событий, обработанных каждой подзадачей через свой пользовательский интерфейс. Однако, некоторые подзадачи перегружены больше других из-за одной из следующих причин:
- фреймворк оптимизирован для большого диапазона ключей, которые будут назначены небольшому количеству контролируемых параллелизмом задач;
- алгоритм назначения подзадач по умолчанию имеет неравномерное распределение.
Для вычисления индекса оператора Apache Flink использует хэш-код ключа и хэш-функцию Murmur для кодирования ключей, а также значения параллелизма: parallelism и maxParallelism, по умолчанию равный 1,5*parallelism. Необходимо указать оба этих значения явно и убедиться, что значение maxParallelism кратно значению parallelism, чтобы получить равномерное распределение. При небольшом количестве ключей, особенно если они представляют собой последовательные целые числа, распределение хэш-функции Murmur по модулю параллелизма также будет неравномерно. Обычно это случается, когда надо перераспределить данные, чтобы каждая подзадача приемника соответствовала одному разделу в нижестоящей системе. В таких случаях удобно перераспределить данные по индексу разделов, чтобы нагрузка была равномерно распределена во Flink. Для этого надо знать схему хэширования Flink и применить небольшой обратный инжиниринг вместо передачи ожидаемого индекса подзадачи в качестве ключа: переназначить индекс каждой подзадачи на целое число idx, чтобы murmur(idx) % количество подзадач == i . Таким образом каждое событие попадет в нужную подзадачу. О том, как еще можно решить проблему перекоса данных во Flink-приложениях, читайте в нашей новой статье.
Работа с состояниями заданий
По умолчанию Flink сохраняет состояние на неопределенный срок: любой ключ, связанный с сеансом будет храниться вечно, если не задано его время хранения в состоянии простоя. В работе с DataStream API приходится фактически контролировать состояние самостоятельно, например, очищать их с помощью таймеров. Также Flink поддерживает время жизни состояние – можно задать значение TTL и очистка будет выполняться автоматически.
В работе с состояниями разработчики Flink-приложений часто используют функции ListState() и ReductionState(). В отличие от ListState(), ReductionState() поддерживает только одну запись за временное окно. Накопительное добавление элементов в список с ListState() в RocksDB, встроенной key-value базе данных, которую использует Flink для хранения состояния, стоит довольно дешево, поскольку не нужно десериализовать существующие записи в состоянии. С ReductionState() значение хранится в сериализованной куче, и каждый раз при обновлении выполняется десериализация, вычисление и сериализация. Это может сильно снизить производительность Flink-приложения.
Впрочем, большое состояние становится проблемой, т.к. это увеличивает накладные расходы на создание контрольной точки – механизма восстановления потокового приложения Apache Flink в случае сбоя. Кроме того, состояние во Flink периодически сохраняется благодаря контрольным точкам, поэтому при сбое его надо восстановить из внешнего хранилища. Если размер состояния большой, это может занять значительное время. Поэтому лучше всего хранить состояния во внешних базах данных. А при работе с RocksDB следует использовать локальные SSD-диски, которые на несколько порядков быстрее, чем стандартные сетевые HDD. Хотя RocksDB хранит некоторые данные в памяти, о чем мы писали здесь, доступ к диску все равно очень важен для быстрого выполнения Flink-заданий.
Отделение хранение данных от потоковых заданий
Хотя передавать данные из задания потоковой передачи непосредственно в хранилище может показаться заманчивым, более надежно использовать промежуточное хранилище типа AWS S3, для разделения систем. В случае проблем с внешним хранилищем, потоковые задания потерпят сбой. Это же эмпирическое правило можно применить к любой интеграции с внешней системой. Если что-то может выйти из строя и может быть недоступно в течение длительного времени, рекомендуется разделить и резервировать ненадежные компоненты. Читайте в нашей новой статье про автоматическое масштабирование Flink-приложений в Googe Cloud Platform.
Освойте тонкости использования Apache Flink и Spark для потоковой обработки событий в распределенных приложениях аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
