Как оптимизатор Calcite в Apache Flink переводит SQL-команды в задания потоковой и пакетной обработки и какие приемы могут ускорить их выполнение. Разбираемся, чем полезны интерфейсы пользовательских коннекторов источника и подсказки запросов.
Flink SQL в пакетной и потоковой обработке данных
Apache Flink позволяет разрабатывать распределенные приложения потоковой обработки больших данных, предоставляя множество полезных библиотек. Одной из них является Flink SQL — реляционный API на основе стандартного языка структурированных запросов. Он обеспечивает унифицированную пакетную и потоковую обработку, предлагая декларативный API относительно высокого уровня. В отличие от API DataStream, чьи примитивы потоковой обработки в низкоуровневом императивном API сразу преобразуются в граф выполнения, программа на Flink SQL должна сперва пройти сложные преобразования. Планировщик SQL предоставляет различные методы и приемы оптимизации. Некоторые из них мы рассмотрим далее.
Напомним, хотя в Apache Flink есть унифицированный API, унифицированное описание оператора и унифицированная структура выполнения, режим выполнения оператора пакетной и потоковой передачи отличается.
Пакетный оператор получает ограниченный набор данных в качестве входа и создает ограниченный набор данных в качестве выхода. Данные будут перенесены на диск, как только закончится память. Поэтому при разработке и реализации пакетных операторов следует избегать доступа к диску для повышения производительности.
Оператор потоковой передачи принимает неограниченный набор данных в качестве входа (лог изменений) и выводит неограниченный набор данных. Поскольку оператор не может буферизовать каждый фрагмент входных данных даже с помощью диска, каждый из них будет обрабатываться оператором индивидуально. Операторы потоковой передачи используют состояние для хранения промежуточных результатов вычислений. Поэтому при разработке и реализации операторов потоковой передачи следует избегать доступа к состоянию для повышения производительности. Лог изменений содержит не только вставку сообщений (SQL-запросы типа INSERT), но и их отзыв сообщения (UPDATE_BEFORE, UPDATE_AFTER, DELETE).
Чтобы код SQL стал эффективным заданием Flink, фреймворк преобразует его в логический план (LogicalPlan), физический план выполнения (ExecPlan), а затем в граф задания (JobGraph), используя оптимизатор. Задание Flink SQL может содержать несколько операторов вставки, которые анализируются и преобразуются в несколько логических деревьев, то есть в направленный ациклический граф (DAG, Directed Acyclic Graph). Оптимизатор DAG принимает LogicalPlan, Flink Conf, Constraints и Statistics в качестве входных данных и генерирует оптимизированный ExecPlan в качестве выходных данных. Подробнее об этом читайте в нашей новой статье про Apache Calcite во Flink SQL.
Flink использует Apache Calcite для оптимизации реляционной алгебры, которая может принимать только одно логическое дерево выражений. Поэтому после того, как оптимизатор получит DAG, он сначала разобьёт DAG на несколько деревьев логических выражений на основе представления, а затем использует Calcite для оптимизации дерева одно за другим. Разделение на основе представлений уменьшит дублирование вычислений. После того, как все деревья выражений реляционной алгебры будут оптимизированы, они будут повторно собраны в DAG физического плана и сократят повторные вычисления с помощью технологии повторного использования подграфов для формирования ExecPlan. Оптимизатор Calcite содержит много правил оптимизации и производных атрибутов, чтобы создать наилучший план выполнения.
Flink SQL использует лексическую политику для идентификаторов (таблиц, атрибутов, имен функций), аналогичную Java. В частности, регистр идентификаторов сохраняется независимо от того, заключены они в кавычки или нет. После этого идентификаторы сопоставляются с учетом регистра. Но, в отличие от Java, обратные галочки позволяют идентификаторам содержать не буквенно-цифровые символы.
Символы Unicode поддерживаются в строковых литералах. Если требуются явные кодовые точки Unicode, можно использовать обратную косую черту (\) в качестве экранирующего символа (по умолчанию) или применить специальный экранирующий оператор UESCAPE, после которого в одинарных кавычках можно написать экранируемый символ.
Эффективный пользовательский коннектор
Flink SQL предоставляет несколько интерфейсов для оптимизации пользовательских коннекторов источника. При этом рекомендуется реализовать следующие интерфейсы:
- SupportsFilterPushDown — планировщик поместит фильтры в коннектор источника таблицы, чтобы избежать чтения недопустимых данных и уменьшить количество операций ввода-вывода при сканировании;
- SupportsProjectionPushDown — планировщик будет передавать необходимые поля в коннектор источника таблицы, чтобы избежать чтения недопустимых столбцов и сократить ввод-вывод при сканировании;
- SupportsPartitionPushDown — планировщик поместит необходимый список разделов в коннектор источника таблицы, чтобы избежать чтения недопустимых разделов и уменьшить количество операций ввода-вывода при сканировании;
- SupportsDynamicFiltering — планировщик передаст данные динамической фильтрации в коннектор источника таблицы, чтобы избежать чтения недопустимых разделов и сократить ввод-вывод при сканировании. В отличие от SupportsPartitionPushDown, фильтрация недопустимых разделов будет происходить во время выполнения, а не на этапе статического планирования.
- SupportsLimitPushDown — планировщик будет передавать информацию о лимите в коннектор источника таблицы, чтобы избежать считывания избыточных данных, для каждого экземпляра источника будет считано только количество записей лимита.
- SupportsAggregatePushDown — планировщик будет передавать агрегатные функции в коннектор источника таблицы, чтобы сократить ввод-вывод при сканировании.
- SupportsStatisticReport — планировщик может получить статистику из соединителя источника таблицы и создать лучший план выполнения.
Flink SQL поддерживает изменение поведения выполнения с помощью подсказок. Подсказки бывают двух видов: табличные подсказки и подсказки запроса.
Табличные подсказки или динамические параметры таблицы позволяют динамически указывать или переопределять ее параметры. Например,
/*+ OPTIONS('lookup.cache'='FULL') */
поможет изменить стратегию кэширования таблицы поиска.
Подсказки запроса полезны, чтобы предложить оптимизатору повлиять на планы выполнения запроса в пределах указанной области запроса. Подсказка LOOKUP используется для изменения поведения оператора соединения поиска. Например, подсказка
/*+ LOOKUP('table'='my_table2', 'async'= 'true') */
позволит включить функцию асинхронного поиска.
Подсказки BROADCAST, SHUFFLE_HASH, SHUFFLE_MERGE и NEST_LOOP используются для выбора стратегии пакетного соединения. Например,
/*+ BROADCAST(t1)*/
позволит предложить оптимизатору выбрать широковещательное хэш-соединение таблиц при реализации SQL-запроса с оператором JOIN.
Освойте тонкости использования Apache Flink и Spark для потоковой обработки событий в распределенных приложениях аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
