dbt Core OpenLineage Integrations

dbt lets you define, test, and document SQL-based transformations, typically against cloud data warehouses, to create reusable, analytics-ready datasets. This page explains dbt Core's key concepts and how to configure OpenLineage so ADOC can track dbt pipeline runs and lineage.

dbt Core concepts

  • Models: SQL files that transform raw data into analytics-ready datasets. Models are the core building block in dbt.

  • Tests: Validate data transformations.

  • Seeds: CSV files loaded into your data warehouse, used as static reference data.

  • Macros: Reusable Jinja templates that promote code reuse across models, support control flow, and enable complex transformations.

How ADOC integrates with dbt

ADOC uses OpenLineage to receive dbt events and build pipelines for lineage tracking and observability. In this pattern, dbt with OpenLineage configured in your environment sends events to ADOC Pipelines, which stores and processes them.

Environment setup

Step 1: Create a Python virtual environment

# create python virtual environment python3 -m venv dbtenv # activate the virtual environment source dbtenv/bin/activate

Step 2: Install required libraries

# install the dbt Core library for Snowflake pip3 install dbt-snowflake # install the OpenLineage library pip3 install openlineage-dbt

Step 3: Clone a sample dbt project (optional)

To follow this guide with a working example, you can clone a public sample project:

git clone https://github.com/mikeharding/jaffle_shop_snowflake.git cd jaffle_shop_snowflake

Configuration

Step 1: Configure the Snowflake connection

Create a profiles.yml file for the dbt-Snowflake connection, and update the details for your environment:

config: use_colors: True jaffle_shop: outputs: dev: account: <your_snowflake_account> database: <your_database> password: <enter_password> role: ACCOUNTADMIN schema: PUBLIC threads: 1 type: snowflake user: <your_user> warehouse: <your_warehouse> target: dev

Step 2: Configure OpenLineage

Add an openlineage.yml configuration file and include the API key from ADOC:

transport: type: http url: <ADOC Control Plane URL> endpoint: /torch-pipeline/api/v1/lineage custom_headers: accessKey: <access_key> secretKey: <secret_key>

Add an environment variable pointing to the configuration file path:

export OPENLINEAGE_CONFIG=<path_to_file>/openlineage.yml

Set up Snowflake

If you're using the sample project, create the initial tables in Snowflake by running the following script in the desired database. Ensure the database name matches the one in profiles.yml:

use database <your_database>; create or replace table raw_customers ( id number, first_name varchar, last_name varchar, email varchar); create or replace table raw_orders ( id number, user_id number, order_date date, status varchar); create or replace table raw_payments ( id number, order_id number, payment_method varchar, amount number);

Run dbt

Step 1: Load demo data

# load demo data dbt seed --profiles-dir <path_to_folder_containing_profiles.yml>

Step 2: Run models with OpenLineage

Use dbt-ol instead of dbt to generate OpenLineage events when running models:

dbt-ol run --profiles-dir <path_to_folder_containing_profiles.yml>

Step 3: Run tests

dbt-ol test --profiles-dir <path_to_folder_containing_profiles.yml>

Once ADOC Pipelines receives the dbt events, it automatically detects the dbt job and creates the lineage.

dbt Core with Airflow

Install OpenLineage on the Airflow worker

Ensure the openlineage-dbt library is installed on the Airflow worker:

pip3 install openlineage-dbt export OPENLINEAGE_CONFIG=<path_to_file>/openlineage.yml

Sample Airflow DAG with a dbt operator

from airflow import DAG from airflow.operators.bash_operator import BashOperator from airflow_dbt.operators.dbt_operator import ( DbtRunOperator, DbtTestOperator, ) from datetime import datetime default_args = { 'start_date': datetime(2022, 1, 1), 'schedule_interval': '@daily' } with DAG('my_dbt_dag', default_args=default_args) as dag: task1 = BashOperator(task_id='task1', bash_command='echo "Task 1"') dbt_run = DbtRunOperator( task_id='dbt_run_models', dir='<complete_path_to_dbt_directory>/dbt_snowflake', profiles_dir='<path_to_folder_containing_profiles.yml>', dbt_bin='dbt-ol', ) task3 = BashOperator(task_id='task3', bash_command='echo "Task 3"') task1 >> dbt_run >> task3

Note: When dbt Core runs through Airflow this way, OpenLineage does not provide complete end-to-end lineage. If your Airflow DAG triggers dbt Cloud jobs (rather than dbt Core), see Link Airflow Tasks to dbt Cloud Pipeline Runs in ADOC instead.

What's next