Выполнение конвейеров агрегации между источниками

Конвейер агрегации из разных источников в Atlas Data Federation выполняет те же этапы агрегации, которые Вы знаете по MongoDB, но данные для каждого этапа могут поступать из разных физических систем — сегмента S3, работающего кластера Atlas или обоих источников одновременно. Объединённый механизм запросов прозрачно обрабатывает маршрутизацию, распределение запросов и объединение результатов. С точки зрения приложения это выглядит как запрос к одной коллекции MongoDB.

Самый простой запрос к разным источникам — это вызов find() для виртуальной коллекции, основанной на файлах S3. Механизм запросов читает файлы, разбирает их и применяет фильтр. Поля фильтра, совпадающие с атрибутами разделов в пути, автоматически приводят к исключению ненужных файлов. Поля, которые не совпадают с атрибутами разделов, применяются как фильтр после чтения данных.

К виртуальным коллекциям на основе S3 можно применять любые этапы агрегации: $match, $group, $project, $sort, $limit. По возможности механизм запросов переносит выполнение этапов ближе к источнику, особенно $match для отсечения разделов и столбцов в Parquet, а остальные этапы выполняет в собственном вычислительном слое после чтения данных.

Самый мощный шаблон работы с разными источниками — использование $lookup для объединения работающей коллекции Atlas с коллекцией, архивированной в S3. Начните конвейер с работающей коллекции (она будет ведущей) и выполните поиск в виртуальной коллекции на основе S3. Всегда размещайте $match в начале конвейера, чтобы свести к минимуму количество выполняемых поисков.

Если в Вашем объединённом экземпляре есть несколько хранилищ кластеров Atlas, Вы можете объединять коллекции из разных кластеров Atlas в одном конвейере. Это полезно для многопользовательских или мультирегиональных развертываний, где данные распределены по отдельным кластерам и требуются отчёты между кластерами без их объединения или создания отдельной базы данных для отчётности.

После выполнения агрегации из разных источников Вы можете записать результаты обратно в работающую коллекцию Atlas с помощью $out или $merge. Это шаблон ETL: прочитать исторические данные из S3, объединить их с актуальными данными, вычислить агрегаты и записать результаты в материализованную коллекцию-представление в Atlas, из которой приложения смогут быстро и недорого читать данные.

При запросе файлов Parquet Data Federation применяет отсечение столбцов: если на этапе $project указаны только определённые поля, механизм запросов читает из файла Parquet только соответствующие столбцы (данные в нём хранятся по столбцам). Это может сократить объём прочитанных данных более чем на 90 % по сравнению с чтением всех столбцов. Размещайте $project как можно раньше в конвейере, чтобы максимально эффективно отсекать столбцы.

Atlas Data Federation записывает подробности выполнения запросов в интерфейсе Atlas на вкладке История запросов. Для каждого запроса отображаются: объём обработанных данных, время выполнения и количество просканированных файлов и разделов. Большой объём обработанных данных обычно означает, что атрибуты разделов отсутствуют или запрос не соответствует ни одному ключу раздела. Используйте этот журнал, чтобы настроить конфигурацию хранилища и шаблоны запросов.

Файлы S3 из разных периодов или систем могут иметь разные схемы: имена полей, типы и структуру. Data Federation корректно обрабатывает такие различия: отсутствующие поля возвращают значение null, а дополнительные поля включаются в результат. В конвейере можно использовать $ifNull, $cond и $convert, чтобы привести различающиеся схемы к единому виду перед группировкой или объединением.

Data Federation не кэширует результаты между запросами — каждый запрос заново читает исходные данные. Для панелей мониторинга, которые регулярно выполняют один и тот же отчёт, используйте шаблон ETL: запланируйте агрегацию, записывающую результаты в коллекцию Atlas с помощью $merge, а затем настройте панель на запросы к быстрой коллекции Atlas. Atlas Triggers могут планировать такое обновление с любым интервалом cron.

Учитывайте текущие ограничения: 1) транзакции не поддерживаются в объединённых экземплярах; 2) использование индексов применяется только к коллекциям на основе Atlas, но не к файлам S3; 3) очень большие наборы результатов могут превысить время ожидания — используйте $out/$merge, чтобы записать результаты, вместо передачи их потоком обратно; 4) задержка выше, чем у запроса к работающему Atlas, из-за операций ввода-вывода S3, поэтому такой подход не подходит для запросов в реальном времени, обращённых непосредственно к пользователям.

В этом уроке Вы узнали, что конвейеры агрегации из разных источников используют те же этапы MongoDB для виртуальных коллекций на основе S3 или кластеров Atlas, $lookup позволяет объединять актуальные данные Atlas с архивами S3 в одном конвейере, а ранний этап $project позволяет отсекать столбцы в файлах Parquet и значительно уменьшать объём просматриваемых данных. Далее мы рассмотрим разбиение данных S3 для повышения производительности запросов.