Contribute Openlineage to `dbt-core` #11688

MassyB · 2025-05-28T15:55:31Z

Resolves #11750

TL;DR

This PR integrates dbt with Openlineage. It unlocks lineage tracking and observability of the dbt pipelines.

Openlineage is an open source standard. From its main page:

OpenLineage is an open platform for collection and analysis of data lineage. It tracks metadata about datasets, jobs, and runs, giving users the information required to identify the root cause of complex issues and understand the impact of changes. OpenLineage contains an open standard for lineage data collection, a metadata repository reference implementation (Marquez), libraries for common languages, and integrations with data pipeline tools.

Openlineage (OL) defines events according to a specification. This PR constructs those OL events by consuming the dbt structured logs and sends them to an endpoint.

The endpoint that consumes OL events is totally configurable by the user. It can be Marquez, Datadog or something else. Examples in this PR are using Datadog.

Problem

Let's build the jaffle shop project using the following command

dbt build

We have this output

root@1805096f0987:/usr/src/dbt# dbt build
12:05:17  Running with dbt=1.9.0
12:05:17  [WARNING]: Deprecated functionality

User config should be moved from the 'config' key in profiles.yml to the 'flags' key in dbt_project.yml.
12:05:18  Registered adapter: postgres=1.9.0
12:05:21  Found 5 models, 1 snapshot, 3 seeds, 23 data tests, 1 source, 434 macros
12:05:21
12:05:21  Concurrency: 2 threads (target='pg')
12:05:21
12:05:22  1 of 32 START seed file public.raw_customers ................................... [RUN]
12:05:22  2 of 32 START seed file public.raw_orders ...................................... [RUN]
12:05:24  1 of 32 OK loaded seed file public.raw_customers ............................... [INSERT 100 in 1.33s]
12:05:24  2 of 32 OK loaded seed file public.raw_orders .................................. [INSERT 99 in 1.34s]
12:05:24  3 of 32 START seed file public.raw_payments .................................... [RUN]
12:05:24  4 of 32 START test source_unique_jaffle_shop_dim_date_id ....................... [RUN]
12:05:24  4 of 32 PASS source_unique_jaffle_shop_dim_date_id ............................. [PASS in 0.65s]
12:05:24  5 of 32 START sql view model public.stg_customers .............................. [RUN]
....
12:05:32  32 of 32 PASS unique_orders_order_id ........................................... [PASS in 0.35s]
12:05:32  31 of 32 PASS relationships_orders_customer_id__customer_id__ref_customers_ .... [PASS in 0.37s]
12:05:32
12:05:32  Finished running 3 seeds, 1 snapshot, 2 table models, 23 data tests, 3 view models in 0 hours 0 minutes and 10.92 seconds (10.92s).
12:05:33  Done. PASS=32 WARN=0 ERROR=0 SKIP=0 TOTAL=32

I've truncated the output but:

We can see when the resource (model, test, snapshot …) started and ended. We see some interleaved logs because the parallelism is set to two threads
We have a summary of the whole execution at the end

This output doesn't tell us the SQL queries executed by every model. We can use the --debug for that:

dbt build --debug

We have the following output:

12:11:28  Running with dbt=1.9.0
12:11:28  running dbt with arguments {'printer_width': '80', 'indirect_selection': 'eager', 'write_json': 'True', 'log_cache_events': 'False', 'partial_parse': 'True', 'cache_selected_only': 'False', 'profiles_dir': '/usr/src/dbt/profiles', 'version_check': 'True', 'warn_error': 'None', 'log_path': '/usr/src/dbt/logs', 'fail_fast': 'False', 'debug': 'True', 'use_colors': 'False', 'use_experimental_parser': 'False', 'no_print': 'None', 'quiet': 'False', 'empty': 'False', 'warn_error_options': 'WarnErrorOptions(include=[], exclude=[])', 'introspect': 'True', 'invocation_command': 'dbt build --debug', 'log_format': 'default', 'target_path': 'None', 'static_parser': 'True', 'send_anonymous_usage_stats': 'False'}
12:11:28  [WARNING]: Deprecated functionality

User config should be moved from the 'config' key in profiles.yml to the 'flags' key in dbt_project.yml.
12:11:29  Registered adapter: postgres=1.9.0
12:11:30  checksum: c99e828bba267739642b5a3ce85f17518764ea526e0e6c4fdc649171c1a66bff, vars: {}, profile: , target: , version: 1.9.0
12:11:31  Partial parsing enabled: 0 files deleted, 0 files added, 0 files changed.
12:11:31  Partial parsing enabled, no changes found, skipping parsing
12:11:32  Wrote artifact WritableManifest to /usr/src/dbt/target/manifest.json
12:11:32  Wrote artifact SemanticManifest to /usr/src/dbt/target/semantic_manifest.json
12:11:33  Found 5 models, 1 snapshot, 3 seeds, 23 data tests, 1 source, 434 macros
12:11:33
12:11:33  Concurrency: 2 threads (target='pg')
12:11:33
12:11:33  Acquiring new postgres connection 'master'
12:11:33  Acquiring new postgres connection 'list_postgres'
12:11:33  Acquiring new postgres connection 'list_postgres'
12:11:33  Using postgres connection "list_postgres"
12:11:33  Using postgres connection "list_postgres"
12:11:33  On list_postgres: /* {"app": "dbt", "dbt_version": "1.9.0", "profile_name": "jaffle_shop", "target_name": "pg", "connection_name": "list_postgres"} */

    select distinct nspname from pg_namespace

12:11:33  On list_postgres: /* {"app": "dbt", "dbt_version": "1.9.0", "profile_name": "jaffle_shop", "target_name": "pg", "connection_name": "list_postgres"} */

    select distinct nspname from pg_namespace

12:11:33  Opening a new connection, currently in state init
12:11:33  Opening a new connection, currently in state init
12:11:33  SQL status: SELECT 11 in 0.136 seconds
12:11:33  SQL status: SELECT 11 in 0.134 seconds
12:11:33  On list_postgres: Close
12:11:33  On list_postgres: Close
12:11:33  Re-using an available connection from the pool (formerly list_postgres, now list_postgres_snapshots)
12:11:33  Re-using an available connection from the pool (formerly list_postgres, now list_postgres_public)
12:11:33  Using postgres connection "list_postgres_snapshots"
12:11:33  Using postgres connection "list_postgres_public"
12:11:33  On list_postgres_snapshots: BEGIN
12:11:33  On list_postgres_public: BEGIN
12:11:33  Opening a new connection, currently in state closed
12:11:33  Opening a new connection, currently in state closed
12:11:33  SQL status: BEGIN in 0.086 seconds
12:11:33  SQL status: BEGIN in 0.084 seconds
12:11:33  Using postgres connection "list_postgres_snapshots"
12:11:33  Using postgres connection "list_postgres_public"
....
12:11:42  On test.jaffle_shop.unique_customers_customer_id.c5af1ff4b1: ROLLBACK
12:11:42  SQL status: SELECT 1 in 0.002 seconds
12:11:42  On test.jaffle_shop.unique_customers_customer_id.c5af1ff4b1: Close
12:11:42  On test.jaffle_shop.accepted_values_orders_status__placed__shipped__completed__return_pending__returned.be6b5b5ec3: ROLLBACK
12:11:42  21 of 32 PASS unique_customers_customer_id ..................................... [PASS in 0.43s]
12:11:42  On test.jaffle_shop.accepted_values_orders_status__placed__shipped__completed__return_pending__returned.be6b5b5ec3: Close
12:11:42  Finished running node test.jaffle_shop.unique_customers_customer_id.c5af1ff4b1
12:11:42  22 of 32 PASS accepted_values_orders_status__placed__shipped__completed__return_pending__returned  [PASS in 0.45s]
12:11:42  Began running node test.jaffle_shop.foo_bar_test_orders_bim.45fc81421f
12:11:42  Finished running node test.jaffle_shop.accepted_values_orders_status__placed__shipped__completed__return_pending__returned.be6b5b5ec3
12:11:42  23 of 32 START test foo_bar_test_orders_bim .................................... [RUN]
12:11:42  Began running node test.jaffle_shop.not_null_orders_amount.106140f9fd
...
12:11:45  Done. PASS=32 WARN=0 ERROR=0 SKIP=0 TOTAL=32
12:11:45  Resource report: {"command_name": "build", "command_success": true, "command_wall_clock_time": 17.469067, "process_in_blocks": "0", "process_kernel_time": 0.741213, "process_mem_max_rss": "176572", "process_out_blocks": "0", "process_user_time": 22.572857}
12:11:45  Command `dbt build` succeeded at 12:11:45.380948 after 17.48 seconds

The SQL queries are interleaved making it hard to read and to follow the execution flow

Observability of the dbt pipeline is not ideal:

For projects that have 100+ resources it becomes difficult to parse the logs
When adding debug and parallelism it becomes even harder to follow the execution even for small projects because logs are interleaved
The execution time of SQL queries is not reported
The lineage (input/output) of models is not reported

This PR is about enhancing the observability of dbt pipelines and addressing the shortcomings mentioned above.

Solution

Instead of relying on the textual logs to report progress of the dbt pipeline, This PR integrates dbt-core with Openlineage. Like what has been done for Apache Airflow.

Below are examples on how we leverage those OL events in Datadog to report on the progression of dbt pipelines.
When running:

dbt run --select orders

In the waterfall view we can see:

The duration of the whole pipeline
The model that has been executed
The sql queries executed in the data platform

This is when we build the entire jaffle shop project:

errors are marked in the specific resource they occurred (in a test below)

An interesting flame graph view when the jaffle shop project is executed using two threads

PR details

You can see a presentation of the integration in this short YT video (relevant part is ~10 minutes long). Be sure to check the linked PRs in order to have more context.

in a nutshell this PR adds a callback that listens for particular dbt structured logs events.
for each of those events an OL event is generated and emitted.

How to test

This PR adds functional tests that checks the generated OL events against expected ones.
You can run them by setting up a dev environment and execute the following command

pytest "tests/functional/openlineage/openlineage_project.py"

If there is a failure you will get a json-path-like to the attribute that has a discrepancy.

For unit tests you can run

pytest  "tests/unit/openlineage/"

Linked PRs/Issues

Additional context form the Openlineage repository

[DBT][Feature] Consume dbt structured logs and report progress in real time OpenLineage/OpenLineage#3314
[Feature][dbt] Consume dbt structured logs for test and build commands OpenLineage/OpenLineage#3362
[INTEGRATION][DBT] Update DBT integration to use Streaming Metadata OpenLineage/OpenLineage#1297
[Feature][dbt] Consume dbt structured logs for test and build commands OpenLineage/OpenLineage#3362
Openlineage TSC meeting, talk on the dbt integration: https://youtu.be/wrKaH4p-ia0?t=481
Openlineage TSC meeting slides

Checklist

I have read the contributing guide and understand what's expected of me.
I have run this code in development, and it appears to resolve the stated issue.
This PR includes tests, or tests are not required or relevant for this PR.
This PR has no interface changes (e.g., macros, CLI, logs, JSON artifacts, config files, adapter interface, etc.) or this PR has already received feedback and approval from Product or DX.
This PR includes type annotations for new and modified functions.