24 октября 2023 года вышел очередной релиз Apache Flink. Знакомимся с главными новинками популярного Big Data фреймворка для разработки потоковых stateful-приложений: JDBC-драйвер для SQL-шлюза, хранимые процедуры для коннекторов, расширенная поддержка SQL, динамическое масштабирование с REST API и RocksDB, улучшение пакетных операций, а также другие полезные фичи Apache Flink 1.18.
Улучшения Flink SQL
Начнем с того, что работа с данными в реляционных БД стала намного проще. Прежде всего, Flink 1.18 поставляется с JDBC-драйвером для шлюза Flink SQL. Поэтому разработчик теперь может использовать любой SQL-клиент, поддерживающий JDBC, для взаимодействия с таблицами реляционной БД через Flink SQL. Это реализовано за счет изменения SQL Gateway — сервиса, который позволяет нескольким удаленным клиентам одновременно выполнять SQL-запросы. Он обеспечивает простой способ отправить задание Flink, найти метаданные и проанализировать данные в Интернете. SQL-шлюз состоит из подключаемых конечных точек и процессора SqlGatewayService, повторно используется конечными точками для обработки запросов. Конечная точка — это точка входа, которая позволяет пользователям подключаться. В зависимости от типа конечных точек пользователи могут использовать разные утилиты для подключения. JDBC-драйвер Flink — это Java-библиотека для подключения и отправки SQL-операторов на SQL Gateway в качестве JDBC-сервера. Перед использованием JDBC-драйвера Flink необходимо запустить SQL Gateway в качестве JDBC-сервера и связать его с кластером Flink.
Следующим значительным улучшением Flink SQL стала поддержка хранимых процедур, которые позволяют упаковать сложную логику манипулирования данными и административные задачи в единый участок кода. Хранимые процедуры работают быстро, могут запускать обработку операций с данными непосредственно во внешней базе, а также с с явным управлением транзакциями и блокировками позволяют лучше обеспечить атомарность операций, безопасность и целостность данных.
Недостатками хранимых процедур считается несовместимость диалектов процедурных языков разных СУБД, что затрудняет смену БД как платформы, а также сложности отладки и поддержки из-за реализации бизнес-логики низкоуровневыми средствами. Тем не менее, эта возможность манипулировать данными напрямую на уровне БД может существенно улучшить производительность системы при ее грамотном применении.
Поэтому поддержка хранимых процедур в Apache Flink SQL улучшает интеграцию с внешними системами. В первую очередь эта возможность предназначена для разработчиков коннекторов Flink, которые теперь могут заранее определять пользовательские хранимые процедуры в соединителях через интерфейс Catalog. Теперь задачи коннектора Flink, которые раньше требовали разработки собственного кода, могут быть заменены простыми вызовами, которые инкапсулируют, стандартизируют и потенциально оптимизируют базовые операции. Пользователи могут выполнять процедуры, используя знакомый CALL-синтаксис, и находить доступные процедуры соединителя с помощью SHOW PROCEDURES. Хранимые процедуры в коннекторах улучшают расширяемость API-интерфейсов SQL и таблиц Flink, а также упрощают доступ к данным и управление ими для пользователей.
Процедуры могут запускать задания FLINK с предоставленным файлом потоковой среды исполнения StreamExecutionEnvironment. Чтобы вызвать хранимую процедуру, ее надо сделать доступной в каталоге, реализовав и вернув с помощью метода Catalog.getProcedure(ObjectPath procedurePath). Класс реализации должен быть публичным, доступным глобально и реализовывать интерфейс org.apache.flink.table.procedures.Procedure. Поскольку интерфейс не предоставляет никакого метода, разработчику необходимо определить публичный метод с именем, который будет принимать четко определенный набор аргументов и реализовывать логику процедуры.
Еще одной новинкой Flink SQL в релизе 1.18 стало расширение поддержки DDL-запросов: теперь можно работать с командами REPLACE TABLE AS SELECT и CREATE OR REPLACE TABLE AS SELECT. Обе эти команды, а также ранее поддерживаемая CREATE TABLE AS теперь обеспечивают атомарность, если сам базовый коннектор также ее поддерживает. Более того, Apache Flink теперь поддерживает TRUNCATE TABLE в режиме пакетного выполнения, когда базовый коннектор имеет эту возможность. Напомним, SQL-оператор TRUNCATE TABLE позволяет пользователям быстро и эффективно очистить, удалив из нее все строки, и не удаляя саму таблицу, в отличие от оператора DROP TABLE. Оператор TRUNCATE TABLE часто используется в хранилищах данных, позволяя повысить производительность за счет сокращения времени и ресурсов при удалении больших объемов данных.
Также в Apache Flink 1.18 добавлена поддержка добавления, удаления и вывода списка разделов с помощью команд ALTER TABLE ADD PARTITION, ALTER TABLE DROP PARTITION и SHOW PARTITIONS.
Еще, начиная с версии 1.18 Flink поддерживает синтаксис SQL для запроса исторических версий данных, позволяя пользователям указывать момент времени, чтобы получать данные и схему таблицы в том виде, в котором они выглядели на тот момент. Путешествуя во времени, пользователи могут легко анализировать и сравнивать исторические версии данных. В стандарте SQL 2011 определен синтаксис, связанный с путешествиями во времени. Его реализация в Apache Flink поддерживает как пакетный режим, так и потоковый режимы.
Также в Apache Flink SQL добавлены runtime-фильтры для оптимизации производительности соединения. Они динамически создают условия фильтрации для определенных запросов соединения во время выполнения, чтобы уменьшить объем сканируемых или перемешиваемых данных, избежать ненужного ввода-вывода и передачи по сети, а также ускорить выполнение запроса. По умолчанию эта функция отключена, ее можно включить с помощью параметра table.optimizer.runtime-filter.enabled.
Наконец, оператор Fusion Codegen повышает производительность выполнения запроса за счет объединения оператора DAG в единый оптимизированный оператор, который исключает вызовы виртуальных функций, использует регистры ЦП для промежуточных данных и уменьшает промахи в кэше инструкций. По умолчанию эта функция также отключена, и ее можно включить с помощью параметра table.exec.operator-fusion-codegen.enabled.
Новинки потоковой и пакетной обработки
Начиная с Flink 1.18, пользователи Table API и SQL могут устанавливать время жизни состояния индивидуально для stateful-операторов. Это полезно в сценариях регулярного соединения потоков, для которых можно устанавливать разные значения TTL (Time To Live). В предыдущих версиях срок действия состояния можно было контролировать только на уровне конвейера с помощью конфигурации table.exec.state.ttl. Благодаря введению сохранения состояния на уровне оператора разработчик может оптимизировать использование ресурсов, сократив выполнение этапа вычислений в задании Flink. Это пригодится, когда требования к сохранению состояния различаются в рамках одного и того же задания, например, отслеживание логистики заказов и вычисление совокупных показателей за каждый день.
Также добавлена возможность настраивать выравнивание водяных знаков и тайм-ауты простоя источника в чистом SQL с помощью подсказок. Раньше эти функции были доступны только в API DataStream. Атрибут времени события определяется с помощью инструкции WATERMARK в таблице CREATE DDL. Оператор WATERMARK определяет выражение создания водяного знака в существующем поле времени события, которое помечает поле времени события как атрибут времени события.
Гибридный shuffle-режим пакетных операций, о котором мы писали здесь, теперь поддерживает хранение данных перемешиваемых в удаленном хранилище, путь к которому можно настроить с помощью параметра taskmanager.network.hybrid-shuffle.remote.path. В версии Flink 1.18 этот режим использует меньше сетевой памяти, чем раньше, за счет ее отделения от параллелизма, что повышает стабильность и простоту использования фреймворка, а также избавляет от нехватки дискового пространства. В облачной среде сложно определить подходящее дисковое пространство для пакетного перемешивания, поскольку обычно локальные диски изолируются на уровне детализации подов Kubernetes. Если пользователи настраивают большой локальный диск пода вручную, его стоимость будет относительно высокой, и затраты будут потрачены впустую, когда диск простаивает. При пользовательской настройке небольшого локального диска пода, крупномасштабные задания могут столкнуться с ошибкой недостаточного дискового пространства при выполнении shuffle-операций, что приведет к снижению стабильности задания. Предотвратить такие ситуации позволяет динамическое переключение между локальным диском и удаленным хранилищем: когда локальный диск заполнен, он автоматически переключается на удаленное хранилище и автоматически переключается обратно, когда на локальном диске снова появляется свободное место.
Динамическое масштабирование
Поддерживая ориентацию на облачное развертывание, Flink-приложения часто работают под управлением Kubernetes, и предполагают гибкую адаптацию к рабочей нагрузке. Такая эластичность, т.е. способность системы адаптироваться к изменениям рабочей нагрузки без прерывания работы, особенно важна для облачных систем, и для длительных потоковых рабочих нагрузок. Адаптивный планировщик был впервые представлен в версии 1.15 и является центральным элементом полностью эластичного развертывания. По своей сути он позволяет заданиям изменять свои требования к ресурсам и параллелизм во время выполнения. Кроме того, он также адаптируется к доступным ресурсам в кластере, изменяя масштаб только тогда, когда кластер может удовлетворить минимально необходимые ресурсы для задания. До версии Flink 1.18 адаптивный планировщик в основном использовался в реактивном режиме, когда одно задание всегда использовало все доступные ресурсы в кластере. В версии 1.18 адаптивный планировщик стал намного мощнее, чтобы в будущем использоваться по умолчанию для рабочих нагрузок потоковой передачи в Apache Flink.
Впрочем, несмотря на базовые возможности адаптивного планировщика, ранее пользователь не мог менять требования к ресурсам задания во время выполнения. В Apache Flink 1.18 можно во время выполнения задания изменить параллелизм любой отдельной задачи задания через веб-интерфейс Flink и REST API. При этом Apache Flink выполняет регулярную операцию масштабирования, как только будут получены необходимые ресурсы для нового параллелизма. Операция изменения масштаба основана не на точке сохранения, а на обычной периодической контрольной точке, которая не создает никаких дополнительных моментальных снимков. Операция масштабирования уже происходит практически мгновенно и с очень коротким временем простоя для заданий с небольшим размером состояния.
Наконец, масштабирование с помощью бэкэнда состояния RocksDB теперь выполняется намного быстрее. Это возможно благодаря параллельной загрузке дескрипторов состояния и параллельной загрузке отдельных файлов. Также отключена упреждающую запись в журнал для пакетной вставки во временные экземпляры RocksDB, которые используются для масштабирования.
В заключение отметим, что Apache Flink 1.18 теперь поддерживает Java 17, а некоторые API фреймворка признаны устаревшими. В частности, коннекторы SourceFunction и SinkFunction будут удалены в Apache Flink 2.0. А вот API DataSet официально признан устаревшим уже сегодня: вместо него рекомендуется использовать DataStream API с пакетным режимом выполнения.
Узнайте больше про использование Apache Flink для потоковой обработки событий в распределенных приложениях аналитики больших данных и машинного обучения на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
