dbt-flink-adapter 1.3.4

The Flink adapter plugin for dbt

DBT-Flink Adapter

This is POC for DBT-FLINK adapter

rerequisities

• Flink 1.16+
• Flink SqlGateway

Usage

Before we start to play with dbt adapter, we need to setup Flink cluster with SqlGateway configured and running. To simplify that process we prepared Docker Compose configuration.

Install dbt-flink-adapter

cd project_example
python3 -m venv venv
source venv/bin/activate
pip install dbt-flink-adapter
dbt --version

dbt-flink should be enlisted among plugins

Configure DBT profile

Locate DBT profile on your machine. It should be in home directory under ~/.dbt/profiles.yml Add there below config:

flink_profile:
  target: dev
  outputs:
    dev:
      type: flink
      host: localhost
      port: 8083
      session_name: test_session

Launch Flink cluster

cd envs/kafka
docker compose up

cd envs/flink-1.16
docker compose up

Play with sample DBT project with dbt-flink adapter

dbt test
dbt run

FLink SQL tables should be created on Flink cluster

tear down Flink cluster

docker compose stop

Creating DBT model

TODO

Source

Type

Flink supports sources in batch and streaming mode, use type to select what execution environment will be used during source creation.

Watermark

To provide watermark pass column and strategy reference under watermark key in config.

Example:

sources:
  - name: my_source
    tables:
      - name: clickstream
        config:
          type: streaming
          connector_properties:
            ...
          watermark:
            column: event_timestamp
            strategy: event_timestamp
        columns:
          - name: event_timestamp
            data_type: TIMESTAMP(3)

SQL passed to Flink will look like:

CREATE TABLE IF NOT EXISTS my_source (
    `<column>` TIMESTAMP(3),
    WATERMARK FOR <column> AS <strategy>
) WITH (
    ...
)

Please refer to Flink documentation about possible strategies: flink-doc/watermark