Подключение Confluent Cloud к dbt v1
Некоторые возможности могут быть ограничены. Чтобы внести свой вклад в разработку, обратитесь к репозиторию исходного кода ниже.
- Поддерживается: Confluent
- Авторы: Confluent
- Репозиторий GitHub: confluentinc/dbt-confluent
- Пакет PyPI:
dbt-confluent
- Канал Slack:
- Поддерживаемая версия dbt: v1.11.0 и новее
- Поддержка dbt: Not Supported
- Минимальная версия платформы данных: n/a
Установка dbt-confluent
Установите адаптер с помощью `pip`:python -m pip install dbt-confluent
Настройка dbt-confluent
Конфигурацию, специфичную для Confluent Cloud, см. на странице настроек Confluent Cloud.
Подключение к Confluent Cloud с помощью dbt-confluent
Используйте адаптер dbt-confluent для подключения к Confluent Cloud for Apache Flink — полностью управляемому сервису обработки потоковых данных. Адаптер развёртывает модели dbt в виде SQL-операторов Flink, которые непрерывно выполняются в Confluent Cloud.
Предварительные требования
- Учётная запись Confluent Cloud
- Вычислительный пул Flink в вашей среде
- Ключ API Flink для сервисной учётной записи с соответствующими разрешениями RBAC
- Python 3.10 или более поздней версии
Установка
Установите dbt-confluent из PyPI:
pip install dbt-confluent
Настройка профиля
Добавьте следующую конфигурацию в файл profiles.yml, чтобы определить целевые среды Confluent Cloud.
my_confluent_project:
target: dev
outputs:
dev:
type: confluent
cloud_provider: aws
cloud_region: us-east-1
organization_id: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
compute_pool_id: lfcp-xxxxx
environment_id: env-xxxxx
dbname: my_kafka_cluster
flink_api_key: "{{ env_var('CONFLUENT_FLINK_API_KEY') }}"
flink_api_secret: "{{ env_var('CONFLUENT_FLINK_API_SECRET') }}"
threads: 1
Описание полей профиля
| Поле | Обязательное | Описание |
|---|---|---|
type | Да | Должно иметь значение confluent. |
cloud_provider | Да | Облачный провайдер вашей среды Confluent Cloud (aws, azure или gcp). |
cloud_region | Да | Облачный регион вашей среды (например, us-east-1, us-west-2). |
organization_id | Да | Идентификатор вашей организации Confluent Cloud (в формате UUID). Его можно найти в консоли Confluent Cloud в разделе Settings. |
compute_pool_id | Да | Идентификатор вычислительного пула Flink для выполнения операторов (например, lfcp-xxxxx). |
environment_id | Да | Идентификатор среды Confluent Cloud (например, env-xxxxx). Соответствует каталогу Flink SQL. |
dbname | Да | Имя кластера Kafka в среде. Соответствует базе данных Flink SQL. |
flink_api_key | Да | Ключ API Flink для аутентификации в Confluent Cloud. Используйте env_var, чтобы не хранить секреты в открытом виде. |
flink_api_secret | Да | Соответствующий секрет API Flink. Используйте env_var, чтобы не хранить секреты в открытом виде. |
execution_mode | Нет | Задаёт режим выполнения операторов по умолчанию. Допустимые значения: streaming_query (по умолчанию), streaming_ddl, snapshot, snapshot_ddl. |
statement_name_prefix | Нет | Добавляет префикс к имени каждого оператора Flink. По умолчанию: dbt-. |
statement_label | Нет | Применяет метку ко всем операторам Flink, чтобы их можно было фильтровать в консоли Confluent Cloud. По умолчанию: dbt-confluent. |
threads | Нет | Задаёт количество моделей, которые dbt выполняет одновременно. По умолчанию: 1. |
Основные понятия Confluent Cloud
В Confluent Cloud for Apache Flink понятия dbt соответствуют понятиям Flink SQL следующим образом:
| Понятие dbt | Понятие Confluent Cloud / Flink SQL |
|---|---|
database (через environment_id) | Каталог Flink SQL (среда Confluent Cloud) |
schema (через dbname) | База данных Flink SQL (кластер Kafka) |
model | SQL-оператор Flink |
table | Топик Kafka со схемой |
Переменные окружения
Используйте переменные окружения для настройки профиля, особенно в конвейерах CI/CD:
export CONFLUENT_FLINK_API_KEY=your-api-key
export CONFLUENT_FLINK_API_SECRET=your-api-secret
Затем ссылайтесь на них в profiles.yml с помощью функции Jinja env_var:
flink_api_key: "{{ env_var('CONFLUENT_FLINK_API_KEY') }}"
flink_api_secret: "{{ env_var('CONFLUENT_FLINK_API_SECRET') }}"
