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 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) 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

• How dbt concepts map to Flink

• Step 1: Install the dbt-confluent adapter

• Step 2: Set up a dbt project

• Step 3: Write dbt models

• Step 4: Run your models

• Step 5: Test your models

• Step 6: Automate with CI/CD

After the walkthrough, you can learn more about:

• Manage pipeline dependencies

• Manage running statements

• Limitations

Prerequisites

You need the following prerequisites to complete this tutorial:

• Access to Confluent Cloud

• A Flink compute pool in your Confluent Cloud environment

• A Flink API key for a service account with appropriate RBAC permissions

• Your Confluent Cloud organization ID, environment ID, and compute pool ID

• Python 3.10+ installed

• dbt Core 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.

Step 1: Install the dbt-confluent adapter

Install the dbt-confluent adapter using pip:

pip install dbt-confluent

Verify the installation:

dbt --version

The output should list confluent as an installed adapter.

Step 2: Set up a dbt project

1. Create a new dbt project:

   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:

   my_flink_project:
     target: dev
     outputs:
       dev:
         type: confluent
         cloud_provider: <your-cloud-provider>
         cloud_region: <your-region>
         organization_id: <your-organization-id>
         environment_id: <your-environment-id>
         compute_pool_id: <your-compute-pool-id>
         flink_api_key: <your-flink-api-key>
         flink_api_secret: <your-flink-api-secret>
         dbname: <your-kafka-cluster-name>
         threads: 1
   

   Tip

   For production deployments, use environment variables instead of hardcoding credentials:

   flink_api_key: "{{ env_var('CONFLUENT_FLINK_API_KEY') }}"
   flink_api_secret: "{{ env_var('CONFLUENT_FLINK_API_SECRET') }}"
   

3. Verify the connection:

   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 endpoint. Set endpoint in place of cloud_provider and cloud_region. Don’t set both.

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.

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:

{{ 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:

{{ config(
    materialized='streaming_table',
    on_schema_drift='ignore'
) }}

The project structure for Flink SQL transformations is:

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 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:

-- 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.

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:

# 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:

SELECT
  `order_id`,
  `customer_id`,
  `price`,
  `$rowtime` AS order_time
FROM {{ source('marketplace', 'orders') }}

Tip

Sources also work with unit test fixtures. Provide mock rows for a source table the same way you would for a ref():

unit_tests:
  - name: test_stg_orders
    model: stg_orders
    given:
      - input: source('marketplace', 'orders')
        rows:
          - order_id: 1
            customer_id: 100
            price: 29.99

Configure materializations in your dbt_project.yml:

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:

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:

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:

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:

dbt run --select stg_orders

To run all models in a specific directory:

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:

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:

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:

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:

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:

-- 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.

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 in your repository settings.

Multi-environment deployments

To deploy across development, staging, and production environments, define separate dbt targets in your profiles.yml:

my_flink_project:
  target: dev
  outputs:
    dev:
      type: confluent
      cloud_provider: <your-cloud-provider>
      cloud_region: <your-region>
      organization_id: <your-organization-id>
      environment_id: <dev-environment-id>
      compute_pool_id: <dev-compute-pool-id>
      flink_api_key: "{{ env_var('DEV_CONFLUENT_FLINK_API_KEY') }}"
      flink_api_secret: "{{ env_var('DEV_CONFLUENT_FLINK_API_SECRET') }}"
      dbname: <dev-kafka-cluster-name>
      threads: 1
    prod:
      type: confluent
      cloud_provider: <your-cloud-provider>
      cloud_region: <your-region>
      organization_id: <your-organization-id>
      environment_id: <prod-environment-id>
      compute_pool_id: <prod-compute-pool-id>
      flink_api_key: "{{ env_var('PROD_CONFLUENT_FLINK_API_KEY') }}"
      flink_api_secret: "{{ env_var('PROD_CONFLUENT_FLINK_API_SECRET') }}"
      dbname: <prod-kafka-cluster-name>
      threads: 4

Run against a specific target:

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:

# 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:

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:

# 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:

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:

-- 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 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.

• 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.

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 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. 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

• Deploy a Flink SQL Statement Using CI/CD and Confluent Cloud for Apache Flink

• Move SQL Statements to Production in Confluent Cloud for Apache Flink

• Schema and Statement Evolution with Confluent Cloud for Apache Flink

• Compute Pools in Confluent Cloud for Apache Flink

• Grant Role-Based Access in Confluent Cloud for Apache Flink

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.