# Deploy Flink SQL Statements with dbt and Confluent Cloud for Apache Flink Deploy and manage Flink SQL transformations in Confluent Cloud for Apache Flink® as [dbt](https://www.getdbt.com/) models, with built-in testing, dependency tracking, and CI/CD support. The `dbt-confluent` adapter runs your Flink SQL statements through standard [dbt (data build tool)](https://www.getdbt.com/) commands, so the same models, tests, and documentation patterns your team uses for batch analytics also apply to streaming transformations on Confluent Cloud. In this walkthrough, you perform the following steps: - [Prerequisites](#flink-deploy-dbt-prerequisites) - [How dbt concepts map to Flink](#flink-deploy-dbt-concepts) - [Step 1: Install the dbt-confluent adapter](#flink-deploy-dbt-install) - [Step 2: Set up a dbt project](#flink-deploy-dbt-project) - [Step 3: Write dbt models](#flink-deploy-dbt-models) - [Step 4: Run your models](#flink-deploy-dbt-run) - [Step 5: Test your models](#flink-deploy-dbt-test) - [Step 6: Automate with CI/CD](#flink-deploy-dbt-cicd) After the walkthrough, you can learn more about: - [Manage pipeline dependencies](#flink-deploy-dbt-dependencies) - [Manage running statements](#flink-deploy-dbt-manage-deployments) - [Limitations](#flink-deploy-dbt-limitations) ## Prerequisites You need the following prerequisites to complete this tutorial: - [Access to Confluent Cloud](https://confluent.cloud/) - A [Flink compute pool](../concepts/compute-pools.md#flink-sql-compute-pools) in your Confluent Cloud environment - A [Flink API key](generate-api-key-for-flink.md#flink-generate-api-key) for a [service account](../../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts) with appropriate [RBAC permissions](flink-rbac.md#flink-rbac) - Your Confluent Cloud organization ID, environment ID, and compute pool ID - [Python 3.10+](https://www.python.org/downloads/) installed - [dbt Core](https://docs.getdbt.com/docs/core/installation-overview) 1.10 or later installed ## How dbt concepts map to Flink Confluent Cloud Flink uses different terminology than the databases that dbt traditionally targets. Before you configure a profile or write models, use the following table to map dbt concepts to Flink SQL and Confluent Cloud terms: | dbt concept | Flink SQL concept | Confluent Cloud entity | |---------------|---------------------|--------------------------| | `database` | Catalog | Environment | | `schema` | Database | Kafka cluster | #### IMPORTANT These mappings determine how you fill in two required fields in your `profiles.yml` connection profile: - `environment_id` is the Confluent Cloud environment ID, in the form `env-xxxxxx`. - `dbname` is the **name** of the Kafka cluster that backs the Flink database, not the cluster ID (`lkc-xxxxxx`). Any model-level `schema` configuration must also reference a cluster by name. The `dbt-confluent` adapter cannot create, rename, or drop Kafka clusters. For more information, see [Limitations](#flink-deploy-dbt-limitations). ## Step 1: Install the dbt-confluent adapter Install the `dbt-confluent` adapter using pip: ```bash pip install dbt-confluent ``` Verify the installation: ```bash dbt --version ``` The output should list `confluent` as an installed adapter. ## Step 2: Set up a dbt project 1. Create a new dbt project: ```bash dbt init my_flink_project ``` When prompted, select `confluent` as the database adapter. 2. Configure your connection profile. Open the `profiles.yml` file (typically at `~/.dbt/profiles.yml`) and add the following configuration: ```yaml my_flink_project: target: dev outputs: dev: type: confluent cloud_provider: cloud_region: organization_id: environment_id: compute_pool_id: flink_api_key: flink_api_secret: dbname: threads: 1 ``` 3. Verify the connection: ```bash dbt debug ``` If the connection is successful, you see an `All checks passed!` message. ### Optional adapter configuration The `dbt-confluent` adapter exposes additional configuration beyond the required connection fields. Some options apply to the whole profile in `profiles.yml`. Others apply to an individual model’s `config()` block. #### Profile options `endpoint` : An alternative to `cloud_provider` and `cloud_region` for connecting through a private or other non-standard cluster URL, such as a [private networking](../concepts/flink-private-networking.md#flink-sql-private-networking) endpoint. Set `endpoint` in place of `cloud_provider` and `cloud_region`. Don’t set both.
```yaml endpoint: https://flink.us-east-2.aws.private.confluent.cloud ``` `statement_name_prefix` : The prefix that the adapter prepends to every generated Flink SQL statement name. Defaults to `dbt-`. The adapter appends your dbt project name and model name after the prefix to build a deterministic statement name.
```yaml statement_name_prefix: my-team- ``` #### Model options Set these in a model’s `config()` block. `statement_name` : Overrides the deterministic statement name that the adapter would otherwise generate for a model. Use this to give a statement a predictable name. You can also point the model at a statement that a previous tool or team already deployed so that `dbt run` adopts and manages it going forward:
```sql {{ config( materialized='streaming_table', statement_name='orders-enriched-insert' ) }} ``` `on_schema_drift` : Controls what happens when a model’s existing Flink SQL table no longer matches its current column list, `WITH` options, or distribution, and the model is re-run without `--full-refresh`. Accepts `fail`, the default, which raises a compile error listing every mismatch, or `ignore`, which skips drift detection entirely:
```sql {{ config( materialized='streaming_table', on_schema_drift='ignore' ) }} ``` The project structure for Flink SQL transformations is: ```text my_flink_project/ ├── dbt_project.yml # Project configuration ├── profiles.yml # Connection profiles (or use ~/.dbt/) ├── models/ │ ├── staging/ # Source-aligned transformations │ │ └── stg_orders.sql │ ├── intermediates/ # Business logic transformations │ │ └── int_order_totals.sql │ └── marts/ # Final output tables │ ├── fct_revenue.sql │ └── schema.yml # Model documentation and tests ├── macros/ # Custom macros ├── tests/ # Custom data tests └── packages.yml # dbt package dependencies ``` ### Supported materializations The `dbt-confluent` adapter supports the following materializations: | Materialization | Description | |--------------------|---------------------------------------------------------------------------------------------------------------------------------------------| | `view` | Creates a Flink SQL view. This is the default materialization. | | `streaming_table` | Creates a streaming table that Flink continuously maintains. Use this
for intermediates and marts that need to be continuously updated. | | `streaming_source` | Creates a source table backed by a connector. Requires a
`connector` config parameter. | #### NOTE The `incremental` materialization is not supported by the `dbt-confluent` adapter. Use `streaming_table` for continuously updated results. #### Create connector-backed sources with `streaming_source` The `streaming_source` materialization creates a Flink SQL table backed by a connector, such as the [faker connector](https://docs.confluent.io/cloud/current/flink/how-to-guides/custom-sample-data.html) for generating test data. Unlike other materializations where the model SQL is a `SELECT` statement, a `streaming_source` model defines the table’s column schema. The connector populates the table with data automatically. For example, a faker source that generates sample order events: ```sql -- models/sources/src_orders.sql {{ config( materialized='streaming_source', connector='faker', with={ 'rows-per-second': '1', 'number-of-rows': '100', } ) }} order_id BIGINT, price DECIMAL(10, 2), order_time TIMESTAMP(3), WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND, PRIMARY KEY(`order_id`) NOT ENFORCED ``` Downstream models can reference the source with `{{ ref('src_orders') }}` like any other model. The `streaming_source` materialization requires `--full-refresh` to replace an existing table. For available connectors and options, see the [Flink SQL CREATE TABLE documentation](https://docs.confluent.io/cloud/current/flink/reference/statements/create-table.html). #### Reference existing Kafka topics with `sources.yml` Use dbt sources when a table already exists in Confluent Cloud Flink and you only need to reference it in a model’s `SELECT` statement. The table might have been created by a connector, produced by another team, or backed by an existing Kafka topic. This is different from `streaming_source`, which is a materialization: running it creates a new table and attaches a connector to populate it. A source creates nothing. It only declares metadata about a table that already exists so that dbt can resolve it by name and track it in the lineage graph. Declare the source in a `sources.yml` file. The source’s `database` corresponds to the Flink SQL catalog (Confluent Cloud environment) that owns the table, per [How dbt concepts map to Flink](#flink-deploy-dbt-concepts): ```yaml # models/sources/sources.yml version: 2 sources: - name: marketplace database: examples tables: - name: orders - name: clicks ``` Reference the source with `{{ source() }}` in a model’s `SELECT` statement: ```sql SELECT `order_id`, `customer_id`, `price`, `$rowtime` AS order_time FROM {{ source('marketplace', 'orders') }} ``` Configure materializations in your `dbt_project.yml`: ```yaml models: my_flink_project: staging: +materialized: view +schema: my_kafka_cluster intermediates: +materialized: streaming_table +schema: my_kafka_cluster marts: +materialized: streaming_table +schema: my_kafka_cluster ``` ## Step 3: Write dbt models Each dbt model corresponds to a Flink SQL statement. Create a model file in the `models/` directory. For example, create `models/staging/stg_orders.sql` to select from an Kafka topic in the `examples` sample catalog: ```sql SELECT `order_id`, `customer_id`, `product_id`, `price`, `$rowtime` AS order_time FROM `examples`.`marketplace`.`orders` ``` Create `models/marts/fct_revenue.sql` to aggregate order data using a tumbling window. Use `{{ ref() }}` to reference other dbt models: ```sql SELECT window_start, window_end, SUM(price) AS total_revenue, COUNT(*) AS order_count FROM TABLE( TUMBLE(TABLE {{ ref('stg_orders') }}, DESCRIPTOR(order_time), INTERVAL '1' MINUTE) ) GROUP BY window_start, window_end ``` #### NOTE For tables in the current catalog and database (set by `environment_id` and `dbname` in your profile), you can use `{{ ref() }}` to reference other dbt models. For source tables in other catalogs, use fully qualified three-part names (`catalog.database.table`) directly in your SQL. Flink SQL uses backtick-quoted identifiers. Use backticks around column names that contain special characters, like the `$rowtime` system column. ## Step 4: Run your models Deploy your Flink SQL statements to Confluent Cloud by running your dbt models: ```bash dbt run ``` This command submits each model as a Flink SQL statement to your compute pool. You can verify that the statements are running in the Cloud Console or with the Confluent CLI. To run a specific model: ```bash dbt run --select stg_orders ``` To run all models in a specific directory: ```bash dbt run --select staging.* ``` ## Step 5: Test your models The `dbt-confluent` adapter supports both **unit tests** for verifying model logic and **data tests** for validating data quality. Run all tests before deploying to production. ### Write unit tests Unit tests validate your model logic by providing mock input data and comparing the output against expected results. Define unit tests in a `schema.yml` file alongside your models. For example, to test the `stg_orders` model, create `models/staging/schema.yml`: ```yaml unit_tests: - name: test_stg_orders model: stg_orders given: - input: source('marketplace', 'orders') rows: - order_id: 1 customer_id: 100 product_id: 10 price: 29.99 order_time: '2024-01-15 10:00:00' - order_id: 2 customer_id: 101 product_id: 11 price: 49.99 order_time: '2024-01-15 10:05:00' expect: rows: - order_id: 1 customer_id: 100 product_id: 10 price: 29.99 order_time: '2024-01-15 10:00:00' - order_id: 2 customer_id: 101 product_id: 11 price: 49.99 order_time: '2024-01-15 10:05:00' ``` Run unit tests: ```bash dbt test --select "test_type:unit" ``` Under the hood, the `dbt-confluent` adapter creates temporary tables on Confluent Cloud using `CREATE TABLE ... LIKE`, inserts the fixture data, runs your model SQL against those tables, and compares the actual output to your expected rows. The adapter cleans up temporary tables automatically after the test completes. ### Write data tests Data tests validate the quality of data produced by your models. Define column tests in a `schema.yml` file: ```yaml models: - name: stg_orders columns: - name: order_id data_type: bigint tests: - not_null - unique - name: price data_type: decimal(10,2) tests: - not_null ``` Run data tests: ```bash dbt test ``` You can also write custom data tests as SQL files in the `tests/` directory. A test passes if the query returns zero rows: ```sql -- tests/assert_positive_prices.sql SELECT order_id, price FROM {{ ref('stg_orders') }} WHERE price <= 0 ``` #### NOTE In Flink SQL streaming mode, `count(*)` over an empty result set returns zero rows instead of one row with value `0`. The `dbt-confluent` adapter handles this automatically for test assertions. ## Step 6: Automate with CI/CD You can integrate dbt with a CI/CD system to automate deployments. The following example shows a GitHub Actions workflow that runs your dbt models when changes are pushed to the main branch. ```yaml on: push: branches: - main jobs: dbt_deploy: name: "Deploy Flink SQL with dbt" runs-on: ubuntu-latest steps: - name: Checkout uses: actions/checkout@v4 - name: Set up Python uses: actions/setup-python@v5 with: python-version: '3.11' - name: Install dependencies run: pip install dbt-confluent - name: Run unit tests run: dbt test --select "test_type:unit" --profiles-dir . env: CONFLUENT_FLINK_API_KEY: ${{ secrets.CONFLUENT_FLINK_API_KEY }} CONFLUENT_FLINK_API_SECRET: ${{ secrets.CONFLUENT_FLINK_API_SECRET }} - name: Deploy models run: dbt run --profiles-dir . env: CONFLUENT_FLINK_API_KEY: ${{ secrets.CONFLUENT_FLINK_API_KEY }} CONFLUENT_FLINK_API_SECRET: ${{ secrets.CONFLUENT_FLINK_API_SECRET }} ``` Store your Flink API key and secret as [GitHub Action Secrets](https://docs.github.com/en/actions/security-for-github-actions/security-guides/using-secrets-in-github-actions) in your repository settings. ### Multi-environment deployments To deploy across development, staging, and production environments, define separate dbt targets in your `profiles.yml`: ```yaml my_flink_project: target: dev outputs: dev: type: confluent cloud_provider: cloud_region: organization_id: environment_id: compute_pool_id: flink_api_key: "{{ env_var('DEV_CONFLUENT_FLINK_API_KEY') }}" flink_api_secret: "{{ env_var('DEV_CONFLUENT_FLINK_API_SECRET') }}" dbname: threads: 1 prod: type: confluent cloud_provider: cloud_region: organization_id: environment_id: compute_pool_id: flink_api_key: "{{ env_var('PROD_CONFLUENT_FLINK_API_KEY') }}" flink_api_secret: "{{ env_var('PROD_CONFLUENT_FLINK_API_SECRET') }}" dbname: threads: 4 ``` Run against a specific target: ```bash dbt run --target prod ``` ## Manage pipeline dependencies In streaming pipelines, Flink SQL statements form a directed acyclic graph (DAG) of dependencies. A fact table depends on intermediate tables, which depend on source tables. dbt manages this automatically through the `{{ ref() }}` function. ### How `ref()` manages deployment order When you use `{{ ref('model_name') }}` in your SQL, dbt: 1. Builds a dependency graph of all models based on their `ref()` calls. 2. Deploys models in topological order, so that upstream models are created before downstream models that depend on them. 3. Resolves the correct fully qualified table name for the target environment. For example, if `fct_revenue` references `stg_orders`, running `dbt run` deploys `stg_orders` first, then `fct_revenue`. ### Understand downstream impact Before changing a model, understand which downstream models depend on it. Use `dbt ls` to list a model and all its downstream dependents: ```bash # List all models downstream of stg_orders dbt ls --select stg_orders+ # List all models upstream of fct_revenue dbt ls --select +fct_revenue # List a model and everything upstream and downstream dbt ls --select +stg_orders+ ``` Use `dbt docs` to visualize your pipeline DAG and explore dependencies interactively: ```bash dbt docs generate dbt docs serve ``` This opens a browser with an interactive lineage graph showing how models connect to each other. Use this to assess the blast radius of changes and plan deployments. ### Selective deployment Deploy only a model and its dependencies using the `--select` flag with graph operators: ```bash # Deploy stg_orders and everything downstream dbt run --select stg_orders+ # Deploy fct_revenue and all its upstream dependencies dbt run --select +fct_revenue # Deploy only models that have changed compared to the last run dbt run --select state:modified --defer --state target/ ``` ## Manage running statements Streaming Flink SQL statements run continuously, so deploying changes requires consideration of statement state and data continuity. ### Understand materialization behavior Each materialization type handles existing relations differently during `dbt run`: | Materialization | Stateful | Redeployment behavior | |--------------------|------------|---------------------------------------------------------------------------------------------------------------| | `view` | No | Drops and recreates the view. No state to preserve. | | `streaming_table` | Yes | Requires `--full-refresh` to redeploy. Without it, `dbt run`
raises an error if the table already exists. | | `streaming_source` | Yes | Requires `--full-refresh` to redeploy. Without it, `dbt run`
raises an error if the table already exists. | Use `--full-refresh` to force redeployment of stateful materializations: ```bash dbt run --full-refresh ``` #### IMPORTANT Redeploying stateful materializations (`streaming_table`, `streaming_source`) drops and recreates the underlying Kafka topics. The operation drops historical data in those topics, and the statement starts processing from the beginning. Plan full refreshes during maintenance windows and coordinate with downstream consumers. ### Configure streaming table options Use the `with` config to set Flink SQL table options on `streaming_table` models: ```sql -- models/intermediates/int_order_totals.sql {{ config( materialized='streaming_table', with={ 'changelog.mode': 'upsert', } ) }} SELECT customer_id, SUM(price) AS total_spent, COUNT(*) AS order_count FROM {{ ref('stg_orders') }} GROUP BY customer_id ``` ### Handle schema evolution When you change the columns in a model’s `SELECT` statement, consider the impact on downstream models and Schema Registry compatibility: - Use `FULL_TRANSITIVE` [schema compatibility](best-practices.md#flink-sql-best-practices-for-statements-compatibility-type) to prevent breaking changes. - Changes to source schemas can require redeploying dependent statements. Use `dbt ls --select model_name+` to identify affected downstream models. - For comprehensive guidance on how schema changes affect running statements, see [Schema and Statement Evolution with Confluent Cloud for Apache Flink](../concepts/schema-statement-evolution.md#flink-sql-schema-and-statement-evolution). - For stateless statements, you can carry over offsets from a previous statement version to avoid reprocessing data. For more information, see [Carry-over Offsets in Confluent Cloud for Apache Flink](carry-over-offsets.md#flink-sql-carry-over-offsets). ## Limitations The `dbt-confluent` adapter has the following limitations: - **No incremental materialization**: Use `streaming_table` for continuously updated results instead. - **No dbt snapshots**: Flink SQL does not support the transaction operations (`MERGE`, `UPDATE` with common table expressions (CTEs)) required for dbt’s [snapshot](https://docs.getdbt.com/docs/build/snapshots) resource type, which tracks Type-2 slowly changing dimensions over time. This is unrelated to a Flink SQL *snapshot* query, a query that runs once and returns a result instead of running continuously. The `dbt-confluent` adapter uses Flink SQL snapshot queries internally, for example to build unit test fixtures. - **No schema management**: A dbt schema maps to a Flink SQL database, which corresponds to an existing Kafka cluster in Confluent Cloud. For more information, see [How dbt concepts map to Flink](#flink-deploy-dbt-concepts). The adapter cannot create, rename, or drop Kafka clusters. Both `dbname` and any model-level `schema` configuration must reference a cluster that already exists. - **No table renames**: Flink SQL does not support `ALTER TABLE` rename. - **Non-transactional**: Confluent Cloud Flink SQL is non-transactional, so partial deployments are possible if a `dbt run` fails midway. ## Related content - [Flink SQL Development Lifecycle in Confluent Cloud for Apache Flink](development-lifecycle.md#flink-development-lifecycle) - [Deploy a Flink SQL Statement Using CI/CD and Confluent Cloud for Apache Flink](../how-to-guides/deploy-flink-sql-statement.md#flink-deploy-sql-statement) - [Move SQL Statements to Production in Confluent Cloud for Apache Flink](best-practices.md#flink-sql-best-practices-for-statements) - [Schema and Statement Evolution with Confluent Cloud for Apache Flink](../concepts/schema-statement-evolution.md#flink-sql-schema-and-statement-evolution) - [Compute Pools in Confluent Cloud for Apache Flink](../concepts/compute-pools.md#flink-sql-compute-pools) - [Grant Role-Based Access in Confluent Cloud for Apache Flink](flink-rbac.md#flink-rbac) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).