Confluent Developer Marketplace Get started
  1. Home
  2. Cloud
  3. Process Data with Confluent Cloud for Apache Flink
  4. Operate and Deploy Confluent Cloud for Apache Flink

Build a Streaming Pipeline with dbt and Confluent Cloud for Apache Flink

The dbt-confluent adapter lets you define Confluent Cloud for Apache Flink® pipelines as dbt models and deploy them with standard dbt commands, so the version control, testing, and documentation workflow your team uses for batch analytics also applies to streaming transformations on Confluent Cloud.

This page is a complete, end-to-end tutorial. You start from an empty directory and finish with a running, tested pipeline that you evolve in place. Every step builds on the previous one, so work through them in order. For every materialization, configuration option, and adapter behavior, see dbt Adapter Reference for Confluent Cloud for Apache Flink.

What you build

You build a four-model order-processing pipeline:

raw_orders                    streaming_source     Faker connector generates order events
 └─ stg_orders                materialized_table   Cleans and stores the orders in a Kafka topic
     ├─ high_value_orders     materialized_table   Orders at or above an amount threshold
     └─ customer_order_stats  materialized_table   Running order count and spend per customer

In this tutorial, you perform the following steps:

Tip

To see a similar pipeline built live, watch Demo: Create streaming data pipelines with the dbt adapter and Materialized Tables.

Prerequisites

You need the following prerequisites to complete this tutorial:

Note

If your organization inspects TLS traffic with a proxy, such as Zscaler, read Certificate errors behind a TLS inspection proxy before you install the adapter.

Step 1: Install the dbt-confluent adapter

  1. Create and activate a Python virtual environment:

    python3 -m venv .venv
source .venv/bin/activate

  2. Install the adapter:

    pip install dbt-confluent

    The adapter installs the version of dbt Core it supports, currently 1.11.x. Don’t install or upgrade dbt-core separately. dbt Core 1.12 is not compatible with the adapter.

  3. Verify the installation:

    dbt --version

    The output lists confluent under Plugins. The output can also say that a newer version of dbt Core is available. Don’t upgrade dbt Core; the adapter supports only the version it installed.

Step 2: Create and connect a dbt project

  1. Create a new dbt project:

    dbt init orders_pipeline

    Select confluent as the adapter, then answer the prompts:

    Prompt

    What to enter

    Backend URL

    Choose Use cloud_provider and cloud_region, then enter your cloud provider, for example aws, and region, for example us-east-2. If you connect over private networking, choose Use a custom endpoint instead. For more information, see Profile configuration.

    organization_id

    Your Confluent Cloud organization ID.

    Compute pool

    Choose Use a specific compute pool and enter your pool ID, or choose the default compute pool option.

    environment_id

    Your environment ID, for example env-xxxxxx.

    dbname

    The name of your Kafka cluster.

    API credentials

    Choose Use a Global API key and enter the key and secret. If you have a Flink API key instead, choose Use a Flink-region API key. The profile then has flink_api_key and flink_api_secret instead of the global key settings, and you skip Step 8.

    execution_mode, statement_name_prefix, statement_label, threads

    Accept the defaults.

    dbt init writes the answers to ~/.dbt/profiles.yml and creates an orders_pipeline project directory.

  2. Move into the project directory and delete the example models that dbt init generates:

    cd orders_pipeline
rm -rf models/example

    The example models select an untyped NULL, which Flink SQL rejects, so dbt run fails if you leave them in place.

  3. Open dbt_project.yml and replace the models: block at the end of the file with the following. This sets a default materialization for each directory you create in the next steps:

    models:
  orders_pipeline:
    raw:
      +materialized: streaming_source
    staging:
      +materialized: materialized_table
    marts:
      +materialized: materialized_table

  4. Open ~/.dbt/profiles.yml. The orders_pipeline profile resembles the following. To keep the secret out of the file, replace it with an env_var() lookup:

    orders_pipeline:
  target: dev
  outputs:
    dev:
      type: confluent
      cloud_provider: aws
      cloud_region: us-east-2
      organization_id: <your-organization-id>
      environment_id: <your-environment-id>
      compute_pool_id: <your-compute-pool-id>
      dbname: <your-kafka-cluster-name>
      global_api_key: <your-global-api-key>
      global_api_secret: "{{ env_var('CONFLUENT_GLOBAL_API_SECRET') }}"
      execution_mode: streaming_query
      statement_name_prefix: dbt-
      statement_label: dbt-confluent
      threads: 1

    If you chose a Flink API key, the profile has flink_api_key and flink_api_secret in place of global_api_key and global_api_secret. If you use an env_var() lookup, export the variable in your shell:

    export CONFLUENT_GLOBAL_API_SECRET=<your-global-api-secret>

  5. Verify the connection:

    dbt debug

    If the connection succeeds, the output ends with All checks passed!. If it fails, see Troubleshooting.

Step 3: Generate sample data with a streaming source

A streaming_source model creates a table with a Flink SQL connector. Instead of a SELECT statement, the model body is the table’s column list. This tutorial uses the faker connector, which generates sample data, so you don’t need a producer.

A Faker table doesn’t store data. Each statement that reads from it gets its own freshly generated rows. In the next step, you store the generated orders in a Kafka topic once, so that every downstream model reads the same orders.

Create models/raw/raw_orders.sql:

{{ config(
    connector='faker',
    with={
        'rows-per-second': '5',
        'fields.order_id.expression': "#{number.numberBetween '1','1000000'}",
        'fields.customer_id.expression': "#{number.numberBetween '1','500'}",
        'fields.product_name.expression': '#{Commerce.productName}',
        'fields.amount.expression': "#{number.randomDouble '2','5','500'}",
        'fields.status.expression': "#{options.option 'PLACED',' shipped ','Delivered','processing'}",
        'fields.event_time.expression': "#{date.past '30','SECONDS'}",
    }
) }}

order_id      BIGINT NOT NULL,
customer_id   BIGINT NOT NULL,
product_name  STRING NOT NULL,
amount        DECIMAL(10, 2) NOT NULL,
status        STRING NOT NULL,
event_time    TIMESTAMP(3) NOT NULL

The status values are deliberately inconsistent in case and whitespace so that the staging model has something to clean. Write single quotes inside with values as-is. The adapter escapes them when it renders the CREATE TABLE statement.

Step 4: Write the transformation models

  1. Create models/staging/stg_orders.sql. A materialized_table model submits a CREATE OR ALTER MATERIALIZED TABLE statement. Flink then runs the query continuously and writes its results to the table’s Kafka topic:

    SELECT
    order_id,
    customer_id,
    product_name,
    LOWER(TRIM(status)) AS status,
    amount,
    event_time
FROM {{ ref('raw_orders') }}
WHERE amount > 0

    {{ ref('raw_orders') }} tells dbt that this model depends on raw_orders, so dbt always deploys raw_orders first.

    Because stg_orders stores the cleaned orders in a topic, both of the following models read the same orders, and Flink can reprocess them later. If stg_orders were a view, each downstream model would read its own separately generated Faker data.

  2. Create models/marts/high_value_orders.sql:

    SELECT
    order_id,
    customer_id,
    product_name,
    amount,
    event_time
FROM {{ ref('stg_orders') }}
WHERE amount >= 300

    This query is stateless: each output row depends only on one input row.

  3. Create models/marts/customer_order_stats.sql:

    {{ config(
    distributed_by={'columns': ['customer_id']},
) }}

SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS lifetime_value
FROM {{ ref('stg_orders') }}
GROUP BY customer_id

    This query is stateful: Flink keeps a running count and total for each customer. The distributed_by configuration partitions the table by customer_id. Distribution columns must come first in the SELECT list.

Your project now contains these models:

orders_pipeline/
├── dbt_project.yml
└── models/
    ├── raw/
    │   └── raw_orders.sql
    ├── staging/
    │   └── stg_orders.sql
    └── marts/
        ├── high_value_orders.sql
        └── customer_order_stats.sql

Step 5: Deploy the pipeline

  1. Deploy every model:

    dbt run

    dbt submits the models in dependency order: raw_orders, then stg_orders, then the two marts. The output reports Completed successfully and PASS=4.

  2. In the Cloud Console, open your Kafka cluster. The stg_orders, high_value_orders, and customer_order_stats topics exist. Open high_value_orders and view its messages. Every order has an amount of at least 300. The first results can take a minute or two to appear.

  3. Run the same command again:

    dbt run

    The second run is safe. The adapter skips raw_orders because its schema hasn’t changed, and re-asserts each materialized table. An unchanged materialized table definition is a no-op in Flink, so no data is dropped or reprocessed.

Step 6: Test the pipeline

dbt supports two kinds of tests. Unit tests check model logic against mock input rows. Data tests check the data that a deployed model produces.

  1. Create models/staging/stg_orders.yml:

    version: 2

models:
  - name: stg_orders
    columns:
      - name: order_id
        data_tests:
          - not_null
      - name: status
        data_tests:
          - not_null
          - accepted_values:
              arguments:
                values: ['placed', 'processing', 'shipped', 'delivered']

unit_tests:
  - name: test_stg_orders_cleans_status_and_drops_invalid_amounts
    model: stg_orders
    given:
      - input: ref('raw_orders')
        rows:
          - {order_id: 1, customer_id: 10, product_name: 'Desk', amount: 350.00, status: ' SHIPPED ', event_time: '2026-01-15 10:00:00'}
          - {order_id: 2, customer_id: 11, product_name: 'Lamp', amount: 0.00, status: 'placed', event_time: '2026-01-15 10:01:00'}
    expect:
      rows:
        - {order_id: 1, customer_id: 10, product_name: 'Desk', amount: 350.00, status: 'shipped', event_time: '2026-01-15 10:00:00'}

    The unit test feeds two mock orders into stg_orders. The test expects the first order’s status to come out lowercase and trimmed, and the second order to be dropped because its amount is zero.

  2. Create a custom data test, tests/assert_high_value_threshold.sql. A custom test passes when its query returns no rows:

    SELECT order_id, amount
FROM {{ ref('high_value_orders') }}
WHERE amount < 300

  3. Run all the tests. Unit tests need the models and their inputs to exist, which is why you deployed the pipeline first:

    dbt test

    To run only the unit tests, run dbt test --select "test_type:unit". For each unit test, the adapter creates temporary tables that copy the schema of the model’s inputs, inserts the mock rows, runs the model’s SQL, and compares the result with the expected rows. It drops the temporary tables when the test finishes.

    Each data test checks every order that is in stg_orders or high_value_orders when the test runs. Orders that arrive later are checked the next time you run dbt test.

Step 7: Evolve a materialized table in place

With a materialized_table model, you change the SQL and run dbt run again. Flink evolves the running table in place, so you don’t drop the table or manage offsets. How the change affects existing results depends on whether the query is stateless or stateful.

Change a stateless model

  1. In models/marts/high_value_orders.sql, lower the threshold and add the status column:

    SELECT
    order_id,
    customer_id,
    product_name,
    amount,
    event_time,
    status
FROM {{ ref('stg_orders') }}
WHERE amount >= 200

  2. Deploy only the changed model:

    dbt run --select high_value_orders

  3. In the high_value_orders topic, new messages include status, and orders between 200 and 300 now appear. Records written before the change stay as they are. The evolved query can take a minute or two to start writing new results.

  4. In tests/assert_high_value_threshold.sql, change 300 to 200 so that the test matches the new logic.

For a stateless query under the default start mode, RESUME_OR_FROM_BEGINNING, Flink resumes from its saved offsets if it can, so the change applies to new records without reprocessing or duplicating earlier ones. For what happens if it can’t, see Evolution and state.

Change a stateful model

  1. In models/marts/customer_order_stats.sql, add a column:

    {{ config(
    distributed_by={'columns': ['customer_id']},
) }}

SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS lifetime_value,
    MAX(amount) AS largest_order
FROM {{ ref('stg_orders') }}
GROUP BY customer_id

  2. Deploy the change:

    dbt run --select customer_order_stats

When a stateful query evolves, Flink discards the query’s state, which here is the running count and total for each customer, and resumes from its saved offsets. When a customer places a new order, Flink overwrites that customer’s row with a count and total that include only orders since the change. Customers with no new orders keep their old rows, which have no largest_order value, so the table mixes old and new results.

To recompute the aggregates from all the orders in the stg_orders topic, rebuild the table with --full-refresh. The rebuilt table has no saved offsets, so the default start mode reads stg_orders from the beginning:

dbt run --select customer_order_stats --full-refresh

Important

--full-refresh drops and recreates the table. This deletes the backing Kafka topic, all of its records, and its Schema Registry schemas, which affects anything that consumes the topic. For more information, see Evolution and state.

Some changes can’t evolve in place. For example, removing a NOT NULL column, such as product_name in high_value_orders, or changing distributed_by fails the run and requires --full-refresh.

Step 8: (Optional) Publish a model with Tableflow

Tableflow materializes the Kafka topic behind a table as an Apache Iceberg™ or Delta Lake table that analytics engines can query. You can enable it on a model with the tableflow configuration. This step requires a global API key in your profile and the Tableflow prerequisites.

  1. Add a config() block to the top of models/marts/high_value_orders.sql:

    {{ config(
    tableflow={
        'table_formats': ['ICEBERG'],
        'storage': {'kind': 'Managed'},
    }
) }}

  2. Deploy the change:

    dbt run --select high_value_orders

  3. In the Cloud Console, open the high_value_orders topic. The Tableflow tab shows Iceberg enabled with Confluent-managed storage.

The adapter checks Tableflow on every run. It enables Tableflow if it’s off and updates the settings in place if you change them. For storage options and limitations, see tableflow.

Step 9: Clean up

The materialized tables run continuously and consume compute pool resources until you remove them.

  1. If you enabled Tableflow in Step 8, disable it on the high_value_orders topic in the Cloud Console.

  2. In a workspace or the Flink SQL shell, with your environment and cluster selected, drop the tables, starting with the downstream ones:

    DROP MATERIALIZED TABLE `customer_order_stats`;
DROP MATERIALIZED TABLE `high_value_orders`;
DROP MATERIALIZED TABLE `stg_orders`;
DROP TABLE `raw_orders`;

Next steps

Related content

Note

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