Интеграция dbt и ClickHouse

ClickHouse Supported

dbt (инструмент построения данных) позволяет аналитическим инженерам преобразовывать данные в своих хранилищах, просто написав операторы select. dbt обрабатывает материализацию этих операторов select в объекты в базе данных в виде таблиц и представлений, выполняя T в Extract Load and Transform (ELT). Пользователи могут создавать модель, определенную оператором SELECT.

Внутри dbt эти модели могут перекрестно ссылаться и структурироваться, позволяя строить более высокоуровневые концепты. Шаблонный SQL, необходимый для подключения моделей, создается автоматически. Более того, dbt идентифицирует зависимости между моделями и обеспечивает их создание в соответствующем порядке с помощью направленного ациклического графа (DAG).

Dbt совместим с ClickHouse через поддерживаемый ClickHouse плагин. Мы описываем процесс подключения ClickHouse с простым примером на основе общедоступного набора данных IMDB. Мы дополнительно выделяем некоторые ограничения текущего коннектора.

• Концепции
• Установка dbt и плагина ClickHouse
• Подключение к ClickHouse
• Создание простой материализации представления
• Создание материализации таблицы
• Создание инкрементного материализованного представления
• Создание снимка
• Использование семян
• Ограничения
• Fivetran
• Связанный контент

Концепции

dbt вводит концепцию модели. Она определяется как оператор SQL, потенциально объединяющий множество таблиц. Модель может быть "материализована" различными способами. Материализация представляет стратегию сборки для запроса select модели. Код, стоящий за материализацией, является шаблонным SQL, который оборачивает ваш оператор SELECT в оператор для создания новой или обновления существующей связи.

dbt предоставляет 4 типа материализации:

• view (по умолчанию): Модель строится как представление в базе данных.
• table: Модель строится как таблица в базе данных.
• ephemeral: Модель не строится непосредственно в базе данных, а вместо этого включается в зависимые модели в виде общих табличных выражений.
• incremental: Модель изначально материализуется как таблица, а при последующих запусках dbt вставляет новые строки и обновляет измененные строки в таблице.

Дополнительный синтаксис и условия определяют, как эти модели должны обновляться, если изменяются их исходные данные. dbt обычно рекомендует начинать с материализации представления, пока производительность не станет проблемой. Материализация таблицы обеспечивает повышение производительности при выполнении запросов, захватывая результаты запроса модели в виде таблицы за счет увеличения памяти. Инкрементальный подход строится на этом дальше, позволяя последующим обновлениям исходных данных захватываться в целевой таблице.

Текущий плагин для ClickHouse поддерживает view, table, ephemeral и incremental материализации. Плагин также поддерживает dbt snapshots и seeds, которые мы рассмотрим в этом руководстве.

В следующих руководствах мы предполагаем, что у вас есть доступ к экземпляру ClickHouse.

Установка dbt и плагина ClickHouse

dbt

Мы предполагаем использование CLI dbt для следующих примеров. Пользователи также могут рассмотреть dbt Cloud, который предлагает веб-ориентированную интегрированную среду разработки (IDE), позволяющую пользователям редактировать и запускать проекты.

dbt предлагает несколько вариантов установки CLI. Следуйте инструкциям, описанным здесь. На этом этапе установите только dbt-core. Мы рекомендуем использовать pip.

Важно: Следующее тестировалось с python 3.9.

Плагин ClickHouse

Установите плагин dbt ClickHouse:

Подготовка ClickHouse

dbt прекрасно справляется с моделированием высоко реляционных данных. Для примера мы предоставляем маленький набор данных IMDB со следующей реляционной схемой. Этот набор данных происходит из репозитория реляционных наборов данных. Это несложно по сравнению с общими схемами, используемыми с dbt, но представляет собой управляемую выборку:

Мы используем подмножество этих таблиц, как показано.

Создайте следующие таблицы:

примечание

Столбец created_at для таблицы roles, который по умолчанию имеет значение now(). Мы используем это позже, чтобы идентифицировать инкрементальные обновления к нашим моделям - см. Инкрементальные модели.

Мы используем функцию s3, чтобы читать исходные данные из общедоступных конечных точек для вставки данных. Выполните следующие команды, чтобы заполнить таблицы:

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

Ответ должен выглядеть следующим образом:

В последующих руководствах мы преобразуем этот запрос в модель - материализуем его в ClickHouse как представление и таблицу dbt.

Подключение к ClickHouse

1. Создайте проект dbt. В данном случае мы назовем его в честь нашего источника imdb. Когда будет предложено, выберите clickhouse в качестве источника базы данных.

2. cd в папку вашего проекта:

3. На этом этапе вам понадобится текстовый редактор по вашему выбору. В нижеследующих примерах мы используем популярный VS Code. Открывая директорию IMDB, вы должны увидеть набор файлов yml и sql:

   

4. Обновите ваш файл dbt_project.yml, чтобы указать нашу первую модель - actor_summary и установить профиль на clickhouse_imdb.

   

5. Далее нам нужно предоставить dbt сведения о подключении к нашему экземпляру ClickHouse. Добавьте следующее в ваш ~/.dbt/profiles.yml.

   Обратите внимание, что необходимо изменить пользователя и пароль. Есть дополнительные доступные параметры, задокументированные здесь.

6. Из директории IMDB выполните команду dbt debug, чтобы подтвердить, что dbt может подключиться к ClickHouse.

   Подтвердите, что ответ включает проверка соединения: [ОК соединение установлено], что указывает на успешное соединение.

Создание простой материализации представления

При использовании материализации представления модель перестраивается как представление при каждом запуске через оператор CREATE VIEW AS в ClickHouse. Это не требует дополнительного хранения данных, но будет медленнее запрашиваться, чем материализации таблицы.

1. Из папки imdb удалите директорию models/example:

2. Создайте новый файл в папке actors внутри папки models. Здесь мы создаем файлы, которые будут представлять модель актера:

3. Создайте файлы schema.yml и actor_summary.sql в папке models/actors.

   Файл schema.yml определяет наши таблицы. Эти таблицы впоследствии будут доступны для использования в макросах. Отредактируйте models/actors/schema.yml, чтобы он содержал этот контент:

   actors_summary.sql определяет нашу фактическую модель. Обратите внимание, что в функции конфигурации мы также запрашиваем, чтобы модель была материализована как представление в ClickHouse. Наши таблицы ссылаются на файл schema.yml через функцию source, например, source('imdb', 'movies') ссылается на таблицу movies в базе данных imdb. Отредактируйте models/actors/actors_summary.sql, чтобы он содержал этот контент:

   Обратите внимание, как мы включаем столбец updated_at в нашу финальную actor_summary. Мы используем это позже для инкрементальных материализаций.

4. Из директории imdb выполните команду dbt run.

5. dbt представит модель как представление в ClickHouse, как и было запрошено. Мы теперь можем напрямую запрашивать это представление. Это представление будет создано в базе данных imdb_dbt - это определяется параметром schema в файле ~/.dbt/profiles.yml под профилем clickhouse_imdb.

   Запрашивая это представление, мы можем воспроизвести результаты нашего более раннего запроса с более простым синтаксисом:

Создание материализации таблицы

В предыдущем примере наша модель была материализована как представление. Хотя это может предложить достаточную производительность для некоторых запросов, более сложные SELECT или часто выполняемые запросы могут быть лучше материализованы как таблица. Эта материализация полезна для моделей, которые будут запрашиваться инструментами BI, чтобы обеспечить пользователям более быстрый опыт. Это фактически приводит к тому, что результаты запроса хранятся как новая таблица с соответствующими накладными расходами на хранение - фактически выполняется INSERT TO SELECT. Обратите внимание, что эта таблица будет перестраиваться каждый раз, то есть она не является инкрементальной. Большие наборы результатов могут, следовательно, привести к длительному времени выполнения - см. Ограничения dbt.

1. Измените файл actors_summary.sql так, чтобы параметр materialized был установлен на table. Обратите внимание, как определен ORDER BY, и обратите внимание, что мы используем движок таблицы MergeTree:

2. Из директории imdb выполните команду dbt run. Это выполнение может занять немного больше времени - около 10 секунд на большинстве машин.

3. Подтвердите создание таблицы imdb_dbt.actor_summary:

   Вы должны увидеть таблицу с соответствующими типами данных:

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

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

Создание инкрементного материализованного представления

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

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

Чтобы проиллюстрировать этот пример, мы добавим актера "Clicky McClickHouse", который появится в невероятных 910 фильмах, убедившись, что он появился в большем количестве фильмов, чем даже Мел Бланк.

1. Сначала мы модифицируем нашу модель, чтобы она была инкрементного типа. Это дополнение требует:

   1. unique_key - Чтобы обеспечить уникальную идентификацию строк, мы должны предоставить unique_key - в данном случае, поле id из нашего запроса будет достаточно. Это гарантирует, что у нас не будет дубликатов строк в нашей материализованной таблице. Для получения дополнительной информации об ограничениях уникальности см. здесь.
   2. Инкрементальный фильтр - Нам также нужно сказать dbt, как он должен выявлять, какие строки изменились за инкрементный запуск. Это достигается путем предоставления выражения дельты. Обычно это включает временную метку для событийных данных; следовательно, наше поле метки времени updated_at. Этот столбец, который по умолчанию принимает значение now() при вставке строк, позволяет идентифицировать новые роли. Кроме того, нам нужно определить альтернативный случай, когда добавляются новые актеры. Используя переменную {{this}}, для обозначения существующей материализованной таблицы, мы получаем выражение where id > (select max(id) from {{ this }}) or updated_at > (select max(updated_at) from {{this}}). Мы встраиваем это внутрь условия {% if is_incremental() %}, обеспечивая его использование только для инкрементных запусков, а не при первом создании таблицы. Для получения дополнительной информации о фильтрации строк для инкрементных моделей см. это обсуждение в документации dbt.

   Обновите файл actor_summary.sql следующим образом:

   Обратите внимание, что наша модель будет реагировать только на обновления и добавления к таблицам roles и actors. Чтобы реагировать на все таблицы, пользователям рекомендуется разделить эту модель на несколько подмоделей — каждая из которых будет иметь свои собственные инкрементные критерии. Эти модели могут в свою очередь быть использованы как ссылки и связываться. Для получения дополнительных сведений о перекрестных ссылках моделей см. здесь.

2. Выполните команду dbt run и подтвердите результаты результирующей таблицы:

3. Теперь мы добавим данные в нашу модель, чтобы проиллюстрировать инкрементное обновление. Добавьте нашего актера "Clicky McClickHouse" в таблицу actors:

4. Пусть "Clicky" сыграет в 910 случайных фильмах:

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

6. Выполните команду dbt run и подтвердите, что наша модель была обновлена и соответствует вышеуказанным результатам:

Внутренности

Мы можем определить выполняемые операторы для выполнения вышеуказанного инкрементного обновления, выполнив запрос к журналу запросов ClickHouse.

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

1. Плагин создает временную таблицу actor_sumary__dbt_tmp. Измененные строки передаются в эту таблицу.
2. Создается новая таблица actor_summary_new. Строки из старой таблицы, в свою очередь, передаются из старой в новую с проверкой, чтобы удостовериться, что идентификаторы строк не существуют во временной таблице. Это эффективно обрабатывает обновления и дубликаты.
3. Результаты из временной таблицы передаются в новую таблицу actor_summary.
4. Наконец, новая таблица атомарно заменяется старой версией с помощью оператора EXCHANGE TABLES. Старая и временные таблицы, в свою очередь, удаляются.

Это визуализировано ниже:

Эта стратегия может столкнуться с проблемами на очень больших моделях. Для получения дополнительной информации см. Ограничения.

Стратегия добавления (режим inserts-only)

Чтобы преодолеть ограничения больших наборов данных в инкрементных моделях, плагин использует параметр конфигурации dbt incremental_strategy. Этот параметр может быть установлен в значение append. При этом обновленные строки вставляются непосредственно в целевую таблицу (также известную как imdb_dbt.actor_summary), и временная таблица не создается. Примечание: режим только добавления требует, чтобы ваши данные были неизменяемыми или чтобы дубликаты были приемлемы. Если вам нужна инкрементная табличная модель, поддерживающая измененные строки, не используйте этот режим!

Чтобы проиллюстрировать этот режим, мы добавим еще одного нового актера и повторно выполним dbt run с incremental_strategy='append'.

1. Настройте режим только добавления в actor_summary.sql:

2. Давайте добавим еще одного известного актера - Дэнни ДеВито.

3. Пусть Дэнни сыграет в 920 случайных фильмах.

4. Выполните dbt run и подтвердите, что Дэнни был добавлен в таблицу actor-summary.

Обратите внимание, как быстро прошло это инкрементное обновление по сравнению с добавлением "Clicky".

Повторная проверка таблицы query_log показывает различия между двумя инкрементными запусками:

В этом запуске только новые строки добавляются непосредственно в таблицу imdb_dbt.actor_summary, и создание таблицы не подразумевается.

Режим Delete+Insert (Экспериментальный)

Исторически ClickHouse имел только ограниченную поддержку обновлений и удалений в форме асинхронных мутаций. Эти мутации могут быть крайне интенсивными по вводу-выведению и их обычно следует избегать.

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

Этот режим можно настроить для модели через параметр incremental_strategy, т.е.

Эта стратегия непосредственно воздействует на таблицу целевой модели, поэтому если возникает проблема во время операции, данные в инкрементной модели, вероятно, окажутся в недействительном состоянии - атомарного обновления не происходит.

Вкратце, этот подход:

1. Плагин создает временную таблицу actor_sumary__dbt_tmp. Строки, которые изменились, передаются в эту таблицу.
2. Осуществляется DELETE в текущую таблицу actor_summary. Строки удаляются по идентификатору из actor_sumary__dbt_tmp.
3. Строки из actor_sumary__dbt_tmp вставляются в actor_summary с использованием команды INSERT INTO actor_summary SELECT * FROM actor_sumary__dbt_tmp.

Этот процесс показан ниже:

Режим insert_overwrite (Экспериментальный)

Выполняет следующие шаги:

1. Создает временную (стейджинговую) таблицу с такой же структурой, как у инкрементной модели: CREATE TABLE {staging} AS {target}.
2. Вставляет только новые записи (полученные с помощью SELECT) в стейджинговую таблицу.
3. Заменяет только новые партиции (присутствующие в стейджинговой таблице) в целевой таблице.

Этот подход имеет следующие преимущества:

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

Создание снимка

Снимки dbt позволяют зафиксировать изменения в изменяемой модели со временем. Это, в свою очередь, позволяет выполнять запросы на определенную дату к моделям, где аналитики могут "оглянуться назад во времени" на предыдущее состояние модели. Это достигается с использованием измерений, меняющихся медленно типа 2, где столбцы "с" и "по" фиксируют, когда строка была действительна. Эта функциональность поддерживается плагином ClickHouse и демонстрируется ниже.

Этот пример предполагает, что вы завершили Создание инкрементной табличной модели. Убедитесь, что ваш actor_summary.sql не устанавливает inserts_only=True. Ваши models/actor_summary.sql должны выглядеть так:

1. Создайте файл actor_summary в директории snapshots.

2. Обновите содержимое файла actor_summary.sql следующим образом:

Несколько замечаний относительно этого содержимого:

• Запрос select определяет результаты, которые вы хотите фиксировать со временем. Функция ref используется для ссылки на нашу ранее созданную модель actor_summary.
• Мы требуем столбец временной метки, чтобы указать изменения записей. Наш столбец updated_at (см. Создание инкрементной табличной модели) может быть использован здесь. Параметр strategy указывает на использование временной метки для обозначения обновлений, с параметром updated_at, указывающим, какой столбец использовать. Если этого столбца нет в вашей модели, вы можете в качестве альтернативы использовать стратегию проверки. Это значительно менее эффективно и требует от пользователя указать список столбцов для сравнения. dbt сравнивает текущие и исторические значения этих столбцов, фиксируя любые изменения (или ничего не делая, если они идентичны).

3. Выполните команду dbt snapshot.

Обратите внимание, что в базе данных snapshots была создана таблица actor_summary_snapshot (определяется параметром target_schema).

4. Делая выборку этих данных, вы увидите, что dbt включил столбцы dbt_valid_from и dbt_valid_to. Последний имеет значения, установленные в null. Последующие запуски будут обновлять это.

5. Сделаем нашего любимого актера Clicky McClickHouse появиться еще в 10 фильмах.

6. Повторно выполните команду dbt run из директории imdb. Это обновит инкрементную модель. После завершения запустите команду dbt snapshot, чтобы зафиксировать изменения.

7. Если мы теперь выполним запрос к нашему снимку, обратите внимание, что у Clicky McClickHouse теперь две строки. Наша предыдущая запись теперь имеет значение dbt_valid_to. Новое значение записывается с тем же значением в столбце dbt_valid_from и значением dbt_valid_to равным null. Если у нас будут новые строки, они также будут добавлены в снимок.

Для получения дополнительной информации о снимках dbt см. здесь.

Использование семян

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

1. Мы генерируем список кодов жанров из нашего существующего набора данных. Из директории dbt используйте clickhouse-client, чтобы создать файл seeds/genre_codes.csv:

2. Выполните команду dbt seed. Это создаст новую таблицу genre_codes в нашей базе данных imdb_dbt (как определено в конфигурации схемы) с записями из нашего файла csv.

3. Подтвердите, что они были загружены:

Ограничения

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

1. Плагин в настоящее время материализует модели как таблицы с использованием INSERT TO SELECT. Это фактически означает дублирование данных. Очень большие наборы данных (PB) могут привести к чрезвычайно долгому времени выполнения, делая некоторые модели нецелесообразными. Стремитесь минимизировать количество строк, возвращаемых любым запросом, используя GROUP BY, когда это возможно. Предпочитайте модели, которые обобщают данные, по сравнению с теми, которые просто выполняют преобразование, сохраняя количество строк источника.
2. Чтобы использовать распределенные таблицы для представления модели, пользователи должны вручную создать базовые реплицируемые таблицы на каждом узле. Распределенная таблица, в свою очередь, может быть создана на основе этих таблиц. Плагин не управляет созданием кластера.
3. Когда dbt создает связь (таблицу/представление) в базе данных, он обычно создает её в формате: {{ database }}.{{ schema }}.{{ table/view id }}. ClickHouse не имеет представления о схемах. Следовательно, плагин использует {{schema}}.{{ table/view id }}, где schema - это база данных ClickHouse.

Дополнительная информация

Предыдущие руководства лишь касаются функциональности dbt. Пользователям рекомендуется ознакомиться с отличной документацией dbt.

Дополнительная конфигурация для плагина описана здесь.

Fivetran

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

Связанный контент

• Блог и вебинар: ClickHouse и dbt - Подарок от сообщества