Перейти к основному содержимому

Настройка CI/CD с помощью пользовательских конвейеров

Назад к гайдам
Платформа dbt
Оркестрация
CI
Средний уровень
Menu

    Введение​

    Один из ключевых принципов dbt заключается в том, что аналитический код должен находиться под контролем версий. Это даёт вашей организации множество преимуществ: удобную совместную работу, единообразие кода, стабильность и возможность отката к предыдущей версии. Платформа хостинга кода предоставляет ещё одно преимущество, которое часто упускают из виду или используют не в полной мере. Некоторые из вас, возможно, уже использовали функциональность вебхуков в dbt для запуска задания при создании PR. Это отличная возможность, которая подходит для большинства сценариев тестирования кода перед его слиянием с кодом рабочей среды. Однако иногда организации нужны дополнительные возможности, например запуск рабочих процессов при каждом коммите (для проверки кода линтером) или после завершения слияния. В этой статье мы покажем, как настроить собственные конвейеры для проверки вашего проекта линтером и запуска задания dbt через API.

    Некоторые из вас, возможно, уже имеют опыт использования функциональности webhook в для запуска job при создании PR. Это отличная возможность, которая покрывает большинство сценариев тестирования кода перед слиянием в production. Однако существуют ситуации, когда организации требуется дополнительная функциональность — например, запуск workflows при каждом коммите (linting) или запуск workflows после завершения слияния.

    В этой статье мы покажем, как настроить кастомные pipelines для линтинга вашего проекта и запуска job в через API.

    Примечание о терминологии в этой статье, так как каждая платформа хостинга кода использует разные термины для схожих концепций. Термины pull request (PR) и merge request (MR) используются взаимозаменяемо, чтобы обозначить процесс слияния одной ветки с другой.

    Что такое конвейеры?​

    Схема работы конвейеровСхема работы конвейеров

    Задания выполняются на раннерах, которые являются виртуальными серверами. Раннеры предварительно настроены с Ubuntu Linux, macOS или Windows. Это означает, что команды, которые вы выполняете, определяются операционной системой вашего раннера. Вы увидите, как это будет использоваться позже в настройке, но пока просто помните, что ваш код выполняется на виртуальных серверах, которые, как правило, размещаются платформой хостинга кода.

    Диаграмма работы конвейеров

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

    • Информация о биллинге для раннеров, размещенных в репозитории:
    • Информация о самостоятелных раннерах:

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

    Предупреждение GitLab о необходимости указать платежные данныеПредупреждение GitLab о необходимости указать платежные данные

    Это руководство предоставляет подробности для нескольких платформ хостинга кода. Где шаги уникальны, они представлены без опции выбора. Если код специфичен для платформы (например, GitHub, GitLab, Bitbucket), вы увидите опцию выбора для каждой.

    Конвейеры могут быть запущены различными событиями. Процесс вебхука dbt Cloud уже запускает выполнение, если вы хотите запускать свои задания на запрос слияния, поэтому это руководство сосредоточено на запуске конвейеров для каждого пуша и при слиянии PR. Поскольку пуши происходят часто в проекте, мы сделаем это задание супер простым и быстрым, используя линтинг с SQLFluff. Конвейер, который запускается на запросах слияния, будет запускаться реже и может быть использован для вызова API dbt Cloud для запуска конкретного задания. Это может быть полезно, если у вас есть специфические требования, которые должны выполняться при обновлении кода в продакшене, например, выполнение --full-refresh на всех затронутых инкрементальных моделях.

    Пайплайны могут запускаться по различным событиям. Механизм вебхуков dbt уже запускает выполнение заданий при создании запроса на слияние, поэтому в этом руководстве основное внимание уделяется запуску пайплайнов при каждом push и при слиянии PR. Поскольку push в проекте выполняются часто, мы сделаем это задание максимально простым и быстрым: оно будет проверять код с помощью SQLFluff. Пайплайн, запускаемый при запросах на слияние, будет выполняться реже, и его можно использовать для вызова API dbt, чтобы запустить определённое задание. Это может быть полезно, если при обновлении кода в production необходимо выполнить особые действия, например запустить --full-refresh для всех затронутых инкрементальных моделей.

    Так как push’и в проекте происходят часто, мы сделаем этот job максимально простым и быстрым, ограничившись линтингом с помощью SQLFluff. Пайплайн, который запускается при merge request, будет выполняться реже и может использоваться для вызова API с целью запуска конкретного job’а. Это может быть полезно, если у вас есть особые требования, которые должны выполняться при обновлении кода в production, например запуск --full-refresh для всех затронутых инкрементальных моделей.

    Диаграмма создаваемых конвейеров и задействованных программДиаграмма создаваемых конвейеров и задействованных программ

    Запуск задания dbt Cloud при слиянии​

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

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

    Если ваш провайдер Git имеет встроенную интеграцию с dbt, вы можете настроить задания слияния в пользовательском интерфейсе.

    :::

    Приведенная ниже настройка показывает, как вызвать API dbt для запуска задания при каждой отправке изменений в вашу ветку по умолчанию.

    1. Получите ваш dbt API key​

    При запуске конвейера CI/CD следует использовать служебный токен вместо API-ключа отдельного пользователя. Этому посвящена подробная документация, а ниже приведена краткая инструкция (эти действия должен выполнять администратор аккаунта):

    1. Войдите в свой аккаунт dbt.
    2. Нажмите на название своего аккаунта в нижней части меню слева и перейдите в Account settings.
    3. Нажмите Service tokens слева.
    4. Нажмите + Create service token, чтобы создать новый токен специально для вызовов API из CI/CD.
    5. Задайте токену имя, например «CICD Token».
    6. Нажмите кнопку +Add permission в разделе Access и предоставьте этому токену разрешение Job Admin.
    7. Нажмите Save, и появится серое поле с вашим токеном. Скопируйте его и сохраните в безопасном месте (это пароль, и обращаться с ним следует соответственно).
    Страница создания сервисных токенов в dbtСтраница создания сервисных токенов в dbt
    Создание сервисного токенаСоздание сервисного токена

    2. Поместите ваш API-ключ dbt в репозиторий​

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

    • Откройте репозиторий, в котором вы хотите запускать пайплайн (тот же, в котором находится ваш проект dbt).
    • Нажмите Settings, чтобы открыть настройки репозитория.
    • Слева, в разделе Security, нажмите раскрывающееся меню Secrets and variables.
    • В открывшемся списке выберите Actions.
    • Ближе к середине экрана нажмите кнопку New repository secret.
    • Вам предложат указать имя — назовём наш секрет DBT_API_KEY.
      • Очень важно скопировать и вставить это имя в точности, поскольку оно используется в скриптах ниже.
    • В поле Secret вставьте ключ, скопированный из dbt.
    • Нажмите Add secret — и всё готово!

    ** Быстрая заметка о безопасности: хотя использование секрета репозитория является самым простым способом настройки этого секрета, в GitHub доступны и другие опции. Они выходят за рамки этого руководства, но могут быть полезны, если вам нужно создать более безопасную среду для выполнения действий. Ознакомьтесь с документацией GitHub о секретах здесь.*

    Вот видео, показывающее эти шаги:

    3. Создайте скрипт для запуска dbt‑задачи через API‑вызов​

    В вашем проекте создайте новую папку на корневом уровне с именем python. В этой папке создайте файл с именем run_and_monitor_dbt_job.py. Скопируйте и вставьте в этот файл содержимое из этого gist.

    my_awesome_project
    ├── python
    │   └── run_and_monitor_dbt_job.py

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

    #------------------------------------------------------------------------------
    # получение переменных окружения
    #------------------------------------------------------------------------------
    api_base        = os.getenv('DBT_URL', 'https://cloud.getdbt.com/') # по умолчанию используется URL многопользовательской версии
    job_cause       = os.getenv('DBT_JOB_CAUSE', 'API-triggered job') # по умолчанию используется общее сообщение
    git_branch      = os.getenv('DBT_JOB_BRANCH', None) # по умолчанию None
    schema_override = os.getenv('DBT_JOB_SCHEMA_OVERRIDE', None) # по умолчанию None
    api_key         = os.environ['DBT_API_KEY']  # здесь нет значения по умолчанию, просто выбросьте ошибку, если ключ не предоставлен
    account_id      = os.environ['DBT_ACCOUNT_ID'] # здесь нет значения по умолчанию, просто выбросьте ошибку, если id не предоставлен
    project_id      = os.environ['DBT_PROJECT_ID'] # здесь нет значения по умолчанию, просто выбросьте ошибку, если id не предоставлен
    job_id          = os.environ['DBT_PR_JOB_ID'] # здесь нет значения по умолчанию, просто выбросьте ошибку, если id не предоставлен

    Требуемый ввод:

    Чтобы вызвать API dbt, скрипту потребуется несколько значений. Проще всего получить их, открыв в dbt задание, которое вы хотите запустить. URL-адрес страницы задания содержит все необходимые значения:

    • DBT_ACCOUNT_ID - это число сразу после accounts/ в URL
    • DBT_PROJECT_ID - это число сразу после projects/ в URL
    • DBT_PR_JOB_ID - это число сразу после jobs/ в URL
    URL задания dbt с выделенными частями, обозначающими аккаунт, проект и заданиеURL задания dbt с выделенными частями, обозначающими аккаунт, проект и задание

    4. Обновите ваш проект, чтобы включить новый вызов API​

    Для этого нового задания добавим файл для вызова API dbt с именем dbt_run_on_merge.yml.

    my_awesome_project
    ├── python
    │   └── run_and_monitor_dbt_job.py
    ├── .github
    │   ├── workflows
    │   │   └── dbt_run_on_merge.yml
    │   │   └── lint_on_push.yml

    Файл YAML будет выглядеть довольно похоже на наше предыдущее задание, но есть новый раздел под названием env, который мы будем использовать для передачи необходимых переменных. Обновите переменные ниже, чтобы они соответствовали вашей настройке на основе комментариев в файле.

    Стоит отметить, что мы изменили секцию on: так, что теперь workflow запускается только при пушах в ветку с именем main (например, когда pull request был смёрджен). Ознакомьтесь с документацией GitHub по этим фильтрам, чтобы узнать о дополнительных сценариях использования.

    Информацию о названиях свойств контекста github и примерах их применения можно найти в документации GitHub.

    name: run dbt job on push

    # Этот фильтр указывает, что задание следует запускать только при отправке изменений в ветку по умолчанию
    # Предполагается, что вы настроили ограничения для этой ветки так, чтобы изменения попадали в неё только через PR
    # Укажите имя, соответствующее имени вашей ветки по умолчанию
    on:
      push:
        branches:
          - 'main'

    jobs:

    # это задание вызывает API dbt для запуска задания
      run_dbt_cloud_job:
        name: Run dbt Job
        runs-on: ubuntu-latest

      # Установите переменные окружения, необходимые для выполнения
        env:
          DBT_ACCOUNT_ID: 00000 # введите ваш id учетной записи
          DBT_PROJECT_ID: 00000 # введите ваш id проекта
          DBT_PR_JOB_ID:  00000 # введите ваш id задания
          DBT_API_KEY: ${{ secrets.DBT_API_KEY }}
          DBT_URL: https://cloud.getdbt.com # enter a URL that matches your job
          DBT_JOB_CAUSE: 'GitHub Pipeline CI Job' 
          DBT_JOB_BRANCH: ${{ github.head_ref }} # Resolves to the head_ref or source branch of the pull request in a workflow run. 

        steps:
          - uses: "actions/checkout@v4"
          - uses: "actions/setup-python@v5"
            with:
              python-version: "3.9"
          - name: Run dbt job
            run: "python python/run_and_monitor_dbt_job.py"

    5. Протестируйте ваше новое действие​

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

    Кроме того, вы увидите это задание в истории запусков dbt. Его будет довольно легко найти: там будет указано, что оно запущено через API, а в разделе INFO будет указана ветка, которую вы использовали в этом руководстве.

    Запуск задания dbt при слиянии в GitHubЗапуск задания dbt при слиянии в GitHub
    Задание dbt, запущенное из GitHubЗадание dbt, запущенное из GitHub

    Запуск задания dbt при pull request​

    Если у вашего Git-провайдера нет встроенной интеграции с dbt, но вы всё же хотите воспользоваться преимуществами CI-сборок, вы обратились по адресу! Приложив немного усилий, можно настроить задание, которое будет запускать задание dbt при создании запроса на слияние (PR).

    Запуск на PR

    Если у вашего Git-провайдера есть встроенная интеграция с dbt, вы можете воспользоваться инструкциями по настройке. Этот раздел предназначен только для проектов, которые подключаются к своему Git-репозиторию с помощью SSH-ключа.

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

    1. Создайте задание конвейера, которое запускается при создании PR​

    Для этой задачи мы настроим запуск с помощью файла bitbucket-pipelines.yml, как и на предыдущем шаге. Этот YAML-файл будет очень похож на файл для предыдущей задачи, но необходимые переменные мы передадим в скрипт Python с помощью команд export. Измените этот раздел в соответствии с вашей конфигурацией, следуя комментариям в файле.

    Что делает этот пайплайн?
    Приведённая ниже конфигурация будет запускать задачу dbt каждый раз, когда в этом репозитории открывается PR. Кроме того, пайплайн будет запускаться заново при каждом новом коммите в PR, пока тот не будет слит.

    Например, если вы откроете PR, запустится пайплайн. Если затем вы решите внести дополнительные изменения и выполните коммит и отправку изменений в ветку PR, запустится новый пайплайн с обновлённым кодом.

    Например: вы открываете PR — pipeline запускается. Затем вы решаете, что нужны дополнительные изменения, и делаете commit/push в ветку PR — будет запущен новый pipeline с обновлённым кодом.

    • DBT_JOB_BRANCH: Указывает заданию dbt выполнять код из ветки, из которой был создан этот PR
    • DBT_JOB_SCHEMA_OVERRIDE: Указывает заданию dbt использовать заданную пользователем целевую схему
      • Формат значения будет следующим: DBT_CLOUD_PR_{REPO_KEY}_{PR_NUMBER}
    image: python:3.11.1


    pipelines:
      # Это задание будет запускаться, когда в репозитории создаются pull requests
      pull-requests:
        '**':
          - step:
              name: 'Run dbt PR Job'
              script:
    # Проверка, позволяющая выполнять сборку, только если целевая ветка PR — master (или другая ветка). 
                # Закомментируйте или удалите строку ниже, если хотите запускать сборку для всех PR независимо от целевой ветки.
                - if [ "${BITBUCKET_PR_DESTINATION_BRANCH}" != "main" ]; then printf ‘PR Destination is not master, exiting.’; exit; fi
                - export DBT_URL="https://cloud.getdbt.com"
                - export DBT_JOB_CAUSE="Bitbucket Pipeline CI Job"
                - export DBT_JOB_BRANCH=$BITBUCKET_BRANCH
                - export DBT_JOB_SCHEMA_OVERRIDE="DBT_CLOUD_PR_"$BITBUCKET_PROJECT_KEY"_"$BITBUCKET_PR_ID
                - export DBT_ACCOUNT_ID=00000 # введите ваш id учетной записи здесь
                - export DBT_PROJECT_ID=00000 # введите ваш id проекта здесь
                - export DBT_PR_JOB_ID=00000 # введите ваш id задания здесь
                - python python/run_and_monitor_dbt_job.py

    2. Подтвердите, что конвейер запускается​

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

    Пайплайн Bitbucket:

    Задание dbt run on PR в BitbucketЗадание dbt run on PR в Bitbucket

    Задание dbt:

    Задание dbt с указанием, что оно запущено из BitbucketЗадание dbt с указанием, что оно запущено из Bitbucket

    3. Обработка этих дополнительных схем в вашей базе данных​

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

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

    • age_in_days: Количество дней с момента последнего изменения схемы, после которого она должна быть удалена (по умолчанию 10 дней)
    • database_to_clean: Имя базы данных, из которой нужно удалить схемы
    {# 
        Этот макрос находит схемы PR старше установленной даты и удаляет их 
        Макрос по умолчанию удаляет схемы старше 10 дней, но может быть настроен с помощью входного аргумента age_in_days
        Пример использования с другой датой:
            dbt run-operation pr_schema_cleanup --args "{'database_to_clean': 'analytics','age_in_days':'15'}"
    #}
    {% macro pr_schema_cleanup(database_to_clean, age_in_days=10) %}

        {% set find_old_schemas %}
            select 
                'drop schema {{ database_to_clean }}.'||schema_name||';'
            from {{ database_to_clean }}.information_schema.schemata
            where
                catalog_name = '{{ database_to_clean | upper }}'
                and schema_name ilike 'DBT_CLOUD_PR%'
                and last_altered <= (current_date() - interval '{{ age_in_days }} days')
        {% endset %}

        {% if execute %}

            {{ log('Schema drop statements:' ,True) }}

            {% set schema_drop_list = run_query(find_old_schemas).columns[0].values() %}

            {% for schema_to_drop in schema_drop_list %}
                {% do run_query(schema_to_drop) %}
                {{ log(schema_to_drop ,True) }}
            {% endfor %}

        {% endif %}

    {% endmacro %}

    Этот макрос добавляется в задание dbt, которое запускается по расписанию. Команда будет выглядеть так (текст для копирования и вставки приведён ниже):

    Задание dbt с командой run-operation для макроса очисткиЗадание dbt с командой run-operation для макроса очистки

    dbt run-operation pr_schema_cleanup --args "{ 'database_to_clean': 'development','age_in_days':15}"

    Учитывайте риск конфликтов при использовании нескольких инструментов оркестрации​

    Запуск заданий dbt через конвейер CI/CD — это один из способов оркестрации заданий. Если вы также запускаете задания с помощью встроенного планировщика dbt, у вас уже два инструмента оркестрации, запускающих задания. Это создаёт риск конфликтов: представьте, что определённые действия запускают конвейер, а одновременно в dbt выполняются задания по расписанию — скорее всего, их запуски будут пересекаться. Чем больше инструментов вы используете, тем больше внимания нужно уделять согласованности их работы.

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

    Нашли ошибку?

    100