В предыдущей статье мы обсудили структуру данных в таблицах Hudi и представили два типа таблиц: CoW и MoR, а также их соответствующие компромиссы. Основываясь на этом, мы теперь рассмотрим, как операции чтения работают в Hudi.

Существуют различные движки, такие как Spark, Presto и Trino, которые интегрируются с Hudi для выполнения аналитических запросов. Хотя API-интерфейсы интеграции могут различаться, основной процесс в механизме распределенных запросов остается прежним. Этот процесс требует интерпретации входного SQL, создания плана запроса, который выполняется на рабочих узлах, и сбора результатов для возврата пользователю.

В этой статье я выбираю Spark в качестве примера механизма, чтобы проиллюстрировать поток операций чтения и предоставить фрагменты кода, демонстрирующие использование различных типов запросов Hudi. Я начну с краткого описания запросов Spark, затем углублюсь в точки интеграции Hudi-Spark и, наконец, объясню различные типы запросов.

Начало работы с запросами Spark

Spark SQL — это распределенный механизм SQL, который может выполнять задачи анализа крупномасштабных данных. Типичный аналитический запрос начинается с предоставленного пользователем SQL-запроса, предназначенного для получения результатов из таблицы в хранилище. Spark SQL принимает эти входные данные и проходит несколько этапов, как показано на изображении ниже.

На этапе анализа входные данные анализируются, анализируются и преобразуются в древовидную структуру как абстракцию оператора SQL. Запросите каталог таблиц, чтобы получить такую ​​информацию, как имена таблиц и типы столбцов.

На этапе логической оптимизации дерево оценивается и оптимизируется на логическом уровне. Некоторые распространенные оптимизации включают перемещение предикатов, сокращение шаблонов и распространение нуля. На этом этапе создается логический план, описывающий вычисления, необходимые для запроса. Поскольку это логическое представление, в логическом плане отсутствуют детали, необходимые для работы на реальных узлах.

Физическое планирование действует как мост между логическим и физическим уровнями. Физический план определяет точный способ выполнения расчетов. Например, в логическом плане может существовать узел соединения, указывающий операцию соединения, тогда как в физическом плане операция соединения может быть указана как соединение сортировки-слияния или соединение широковещательного хеширования, в зависимости от оценки размера соответствующую таблицу. Выберите лучший физический план для генерации кода и фактического выполнения.

Эти три этапа представляют собой функции, предоставляемые Catalyst Optimizer[1]. Чтобы глубже изучить эту тему, вы можете изучить отличные доклады, ссылки на которые можно найти здесь [2] и здесь [3].

Во время выполнения приложения Spark работают на базовой структуре данных, называемой RDD (устойчивый распределенный набор данных). RDD — это коллекции объектов JVM, которые являются неизменяемыми, разделенными по узлам и отказоустойчивыми благодаря отслеживанию информации о происхождении данных. При запуске приложения выполняются запланированные вычисления: преобразуются RDD и выполняются операции для получения результатов. Этот процесс также часто называют «материализацией» СДР.

API источника данных

Когда Catalyst Optimizer разрабатывает план запроса, становится выгодным подключиться к источнику данных, чтобы снизить оптимизацию. API DataSource Spark предназначен для обеспечения расширяемости для интеграции с различными источниками данных. Некоторые источники доступны «из коробки», например JDBC, таблицы Hive и файлы Parquet. Таблицы Hudi представляют собой еще один тип настраиваемого источника данных из-за их специфического расположения данных.

Процесс чтения Спарк-Худи

На следующем рисунке показаны некоторые ключевые интерфейсы и вызовы методов в процессе чтения Spark-Hudi.

1. 1. DefaultSource как интегрирует точку входа, определяя источник данных в формате как org.apache.hudi или hudi。это обеспечивает BaseRelation Реализация, я рассматриваю это как установление «отношений» для упрощения доступа к таблице.
2. 2. buildScan() — это API-интерфейс для передачи фильтров источнику данных для оптимизации. Худи определяет метод CollectFileSplits() для сбора связанных файлов.
3. 3. collectFileSplits() Передать все фильтры FileIndex объект,Долженобъектполезныйидентифицироватьчитатьиз Необходимые документы。
4. 4. FileIndex находит все связанные FileSlices для дальнейшей обработки.
5. 5. Вызовите функцию composeRDD() после идентификации FileSlices.
6. 6. FileSlice загружается и читается как RDD. Для столбчатых файлов, таких как базовые файлы в Parquet, эта операция чтения минимизирует передаваемые байты, считывая только необходимые столбцы.
7. 7. RDD возвращается из API для дальнейшего планирования и генерации кода.

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

Этот процесс является общим для всех типов запросов Spark Hudi. В следующих разделах объясняется, как работают различные типы запросов. Все это применимо к таблицам CoW и MoR, за исключением оптимизации чтения.

запрос моментального снимка

Это тип запроса по умолчанию при чтении таблиц Hudi. Его цель — получить последние записи из таблицы, по сути, сделав «снимок» таблицы на момент запроса. При выполнении таблиц MoR происходит объединение файлов журнала с базовыми файлами, что приводит к некоторому снижению производительности.

Эти SQL-коды можно запустить после запуска оболочки Spark SQL с зависимостью Hudi, чтобы настроить таблицу MoR со вставленной и обновленной записью.

create table hudi_mor_example (
  id int,
  name string,
  price double,
  ts bigint
) using hudi
tblproperties (
  type = 'mor',
  primaryKey = 'id',
  preCombineField = 'ts'
) location '/tmp/hudi_mor_example';

set hoodie.spark.sql.insert.into.operation=UPSERT;
insert into hudi_mor_example select 1, 'foo', 10, 1000;
insert into hudi_mor_example select 1, 'foo', 20, 2000;
insert into hudi_mor_example select 1, 'foo', 30, 3000;

Это можно сделать, запустив, как показано ниже. SELECT оператор выполнения запроса моментального снимок, он получит запись по последнему значению.

spark-sql> select id, name, price, ts from hudi_mor_example;
1       foo     30.0    3000
Time taken: 0.161 seconds, Fetched 1 row(s)

Запросы, оптимизированные для чтения (RO)

Тип запроса RO предназначен для достижения компромисса между меньшей задержкой чтения и потенциально более старыми результатами, поэтому он разработан специально для использования с таблицами MoR. При выполнении таких запросов метод CollectFileSplits() будет получать только FileSlices базовых файлов (файлов Parquet).

Приведенный выше код установки автоматически создает файл с именем hudi_mor_example_ro таблица каталогов, определяющая атрибуты hoodie.query.as.ro.table=true。Этот атрибут указывает Запрос Движок всегда выполняет RO Запрос. Запустите следующее SELECT Этот оператор возвращает исходное значение записи, поскольку последующие обновления еще не были применены к базовому файлу.

spark-sql> select id, name, price, ts from hudi_mor_example_ro;
1       foo     10.0    1000
Time taken: 0.114 seconds, Fetched 1 row(s)

запрос о путешествии во времени

Указав временную метку, пользователи могут запросить исторический снимок таблицы Hudi в определенный момент времени. Как упоминалось выше 1 В этой статье обсуждается FileSlices. Связано с конкретным временем отправки, что поддерживает фильтрацию. выполнить запрос запроса о путешествии во времени, если точного совпадения нет, FileIndex Найти только те, которые соответствуют указанному времени или произошли раньше него. FileSlice。

spark-sql> select id, name, price, ts from hudi_mor_example timestamp as of '20230905221619987';
1       foo     30.0    3000
Time taken: 0.274 seconds, Fetched 1 row(s)

spark-sql> select id, name, price, ts from hudi_mor_example timestamp as of '20230905221619986';
1       foo     20.0    2000
Time taken: 0.241 seconds, Fetched 1 row(s)

первый SELECT утверждение точно в последнем вставленном deltacommit Запрос на выполнение времени о путешествии во времени,Предоставляет последний снимок таблицы. Вторая настройка Запроса с меткой времени предшествует последней вставке метки времени.,и сгенерируйте предпоследнюю вставку из снимка.

Временные метки в примере соответствуют формату временной шкалы Hudi «гггг ММдд ЧЧммсс ССС». Его также можно задать в форме «гггг-ММ-дд ЧЧ:мм:сс.ССС» или «гггг-ММ-дд».

инкрементный запрос

Пользователи могут установить временную отметку начала (с конечной отметкой или без нее) для получения записей, которые изменились в течение определенного временного окна. Если время окончания не установлено, временное окно будет включать самую последнюю запись. Hudi также обеспечивает функцию полного захвата измененных данных (CDC), включая дополнительное ведение журнала на стороне записи и активируя режим CDC для инкрементных считывателей. Более подробная информация будет рассмотрена в отдельном посте, посвященном инкрементальной обработке.

обзор

В этой статье мы даем обзор оптимизатора Catalyst Spark, изучаем, как Hudi реализует API Spark DataSource для чтения данных, а также представляем четыре различных типа запросов Hudi. Фрагменты кода также можно найти здесь [4]. В следующей статье демонстрация будет включена в процесс дальнейшего углубления нашего понимания Худи.