Временное хранилище данных на Apache Druid: почему это эффективно сработало для загрузки табличных файлов

Всем привет! Меня зовут Амир, я Data Engineer в компании «ДЮК Технологии». Подготовил кейс, как мы реализовали для одного крупного заказчика проект по созданию хранилища разрозненных табличных данных.

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

В статье расскажу, почему для реализации проекта мы выбрали именно Druid, с какими особенностями реализации столкнулись, как сравнивали методы реализации датасорсов.

Задача — создать временное хранилище данных (ВХД), чтобы хранить в базах данных загруженные туда JSON/CSV/XLS/XLSX. Затем на основе данных создавать аналитические витрины.

Требуемые характеристики системы:

Инструментом для хранения данных из файлов был выбран Apache Druid — это колоночная база данных реального времени, хорошо подходящая для быстрой обработки больших, редко изменяющихся массивов данных и немедленного предоставления доступа к ним.

Apache Druid имеет уникальную распределенную и эластичную архитектуру, которая предварительно собирает данные из общего слоя данных в практически бесконечный кластер серверов данных. Эта архитектура обеспечивает более высокую производительность и большую масштабируемость по сравнению с такими базами данных, как PostgreSQL и MySQL.

Apache Druid хорошо подходит для создания графических пользовательских интерфейсов (GUI) аналитических приложений. Он также хорошо вписывается в бэк-энд высокопоточных API, которым требуется быстрое агрегирование данных.

Apache Druid поддерживает множество методов и источников для получения и хранения данных и предоставляет большой выбор встроенных API для выполнения тех или иных действий с данными и самой БД, что сильно облегчает взаимодействие с базой данных при помощи внешних инструментов и создания собственного UI с бэк-эндом.

Инструмент Apache Druid используют Airbnb, British Telecom, Paypal, Salesforce, Netflix, Cisco.

Каждая компания использует Apache Druid уникальным образом:

Теперь немного об устройстве хранения данных в Druid. Это пригодится для того, чтобы в дальнейшем понимать трудности, с которыми придется столкнуться.

На макро-уровне все данные, хранимые БД Druid, располагаются в Базах данных, которые в свою очередь разбиваются на датасорсы (datasource), которые по смыслу и строению ближе всего к «таблице» из реляционных баз данных, однако способ их записи на диск отличается из-за ориентированности на большие данные и хранение записей поколоночно. Каждый датасорс может иметь свою собственную схему, правила загрузки, ротации и прочее.

Каждая запись в датасорсе обязана иметь метку времени. С помощью неё данные разделяются на временные периоды, называемые временным отрезком (time chunk). Так в дизайн системы закладывается принцип партиционирования. Также, помимо времени, данные можно дополнительно разделять с помощью иных атрибутов, но это не обязательно.

Временные отрезки делятся на сегменты (segment), которые и записываются в постоянное хранилище.

Сегмент — это архив, содержащий несколько специальных файлов, в которых сохранена какая-то часть датасорса (datasource). Данные в нём записываются поколоночно.

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

Работали с Apache Druid версии 24.0.1.

Для реализации наших целей мы создали веб-интерфейс, который подключался и отправлял в Apache Druid команды, используя его API.

Для загрузки пользовательских данных в Apache Druid мы использовали встроенный пакетный ввод (Native batch ingestion) — метод, при котором нужно загрузить в Druid так называемую спецификацию задания на загрузку (ingestion task spec, далее its), в которой указываются метаданные о загружаемом файле (местонахождение файла, формат файла, выгружаемые колонки, градация и прочее).

Спецификация позволяет быстро и эффективно загружать данные в Druid, используя заданные параметры и настройки. Большое количество параметров, которые можно настроить, помогает тонко настраивать индексацию данных и получать оптимальную производительность при запросах в Druid.

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

Ниже представлены основные параметры, которые могут использоваться в Ingestion spec:

В качестве технологий для бэкэнда мы выбрали Python и PostgreSQL. Далее расскажу, почему и для каких целей.

Apache Druid хранит метаданные необходимые для работы кластера во внешних хранилищах, а внутри себя хранит только фактические данные, загружаемые пользователем. Для хранения метаданных поддерживаются Derby, MySQL и PostgreSQL. Мы выбрали PostgreSQL, так как это стабильное и гибкое open source решение, широко применяемое во многих крупных проектах.

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

Мы использовали Python в качестве инструмента для написания бэк-энд части нашего веб-интерфейса, в частности использовали библиотеку FastAPI для написания микросервисов вызываемых нашим фронт-эндом.

Python хорошо подошел для решения наших задач, так как имеет все готовые библиотеки для взаимодействия с PostgreSQL и для вызова и обработки сторонних API.

Нами были написаны несколько микросервисов, с помощью которых веб-интерфейс мог взаимодействовать с PostgreSQL и Druid, создан функционал загрузки файлов на сервер и последующей их обработкой перед загрузкой в Apache Druid.

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

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

Необходимо было создать гибкий шаблон its, который можно было подстраивать на уровне бэк-энда в момент загрузки файлов на сервер.

Для этого мы создали таблицу в PostgreSQL, куда сохранили шаблоны its с переменными вместо значений для параметров и написали функцию, которая получала от Python нужные значения для параметров, вычитывая данные из файла и подставляла в шаблон необходимые метаданные.

Следующим сложным шагом было решение проблемы хранения данных в Druid, а именно создание для каждого нового файла отдельный датасорс либо же создание для каждого пользователя отдельный датасорс, в который будут загружаться все его файлы. И мы провели исследование плюсов и минусов обоих этих вариантов, один файл — один датасорс (ОФОД), и один пользователь — один датасорс (ОПОД).

У каждого из двух вариантов были свои недостатки и положительные стороны.

По совокупности плюсов и минусов приняли решение для первоначальной реализации использовать метод ОФОД.

Для четкого разделения датасорсов между пользователями решили, что у каждого пользователя название датасорса будет начинаться с префиксом «логин+.+название загруженного файла», и каждый пользователь будет иметь доступ только к своим датасорсам.

Также нам нужно было решить вопрос хранения файлов, загруженных через веб-интерфейс, и дальнейшего их удаления после того, как данные этих файлов будут загружены в Apache Druid.

Было принято решение создать на сервере директорию, в которой эти файлы будут храниться и удаляться ежедневно по расписанию.

Стандартная схема работы Apache Druid:

В использованной нами версии Apache Druid не поддерживались DDL команды по типу create table/view, вместо этого можно было создавать модифицированные копии существующих датасетов с помощью конструкций INSERT-SELECT или REPLACE- SELECT.

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

В результате внедрения мы получили рабочий продукт:

Какие есть планы развития продукта: