Integrating Snowflake + dbt with Dagster Cloud Insights (Experimental)#

External metrics, such as Snowflake credits, can be integrated into the Dagster Insights UI. The dagster-cloud package contains utilities for capturing and submitting external metrics about data operations to Dagster Cloud via an API.

If you use dbt to materialize tables in Snowflake, use this guide to integrate Snowflake metrics into the Insights UI. For instructions on integrating direct Snowflake queries, see Integrating Direct Snowflake Usage with Dagster Cloud Insights.

Prerequisites#

To complete the steps in this guide, you'll need:

• A Dagster Cloud account on the Pro plan

• Access to the Dagster Cloud Insights feature

• Snowflake credentials which have access to the snowflake.account_usage.query_history table. For more information on granting access to this table, see the Snowflake documentation.

• To install the following libraries:

  pip install dagster dagster-cloud dagster-dbt dagster-snowflake
  

  Note: If you already have dagster-cloud installed, make sure you're using version 1.5.1 or newer.

Step 1: Instrument your Dagster code#

from dagster_cloud.dagster_insights import dbt_with_snowflake_insights


@dbt_assets(...)
def my_asset(context: AssetExecutionContext, dbt: DbtCliResource):
    # Typically you have a `yield from dbt_resource.cli(...)`.
    # Wrap the original call with `dbt_with_snowflake_insights` as below.
    dbt_cli_invocation = dbt_resource.cli(["build"], context=context)
    yield from dbt_with_snowflake_insights(context, dbt_cli_invocation)

from dagster_cloud.dagster_insights import dbt_with_snowflake_insights


@op(out={})
def my_dbt_op(context: OpExecutionContext, dbt: DbtCliResource):
    # Typically you have a `yield from dbt_resource.cli(...)`.
    # Wrap the original call with `dbt_with_snowflake_insights` as below.
    dbt_cli_invocation = dbt.cli(
        ["build"], context=context, manifest=dbt_manifest_path
    )
    yield from dbt_with_snowflake_insights(context, dbt_cli_invocation)

@job
def my_dbt_job():
  ...
  my_dbt_op()
  ...

Step 2: Update dbt_project.yml#

Next, add the following to your dbt project's dbt_project.yml:

query-comment:
  comment: "snowflake_dagster_dbt_v1_opaque_id[[[{{ node.unique_id }}:{{ invocation_id }}]]]"
  append: true

This allows you to add a comment to every query recorded in Snowflake's query_history table. The comments will contain the dbt invocation ID and unique ID.

Note: Make sure to include append: true, as Snowflake strips leading comments.

Step 3: Create a metrics ingestion pipeline in Dagster#

The last step is to create a Dagster pipeline that joins asset observation events with the Snowflake query history and calls the Dagster Cloud ingestion API. Snowflake usage information is available at a delay, so this pipeline will run on a schedule to ingest Snowflake usage information from the previous hour.

Note that you only need to create this pipeline in a single code location per deployment, even if you have instrumented dbt assets in multiple code locations.

To do this, you'll need a Snowflake resource (SnowflakeResource) that can query the snowflake.account_usage.query_history table. You can set up the ingestion pipeline like the following:

from dagster_snowflake import SnowflakeResource
from dagster import Definition, EnvVar

from dagster_cloud.dagster_insights import (
    create_snowflake_insights_asset_and_schedule,
)

snowflake_insights_definitions = create_snowflake_insights_asset_and_schedule(
    start_date="2023-10-5-00:00",
    snowflake_resource_key="snowflake_insights",
)

defs = Definitions(
  assets=[..., *snowflake_insights_definitions.assets],
  schedules=[..., snowflake_insights_definitions.schedule],
  resources={
    ...,
    "snowflake_insights": SnowflakeResource(
      account=EnvVar("SNOWFLAKE_PURINA_ACCOUNT"),
      user=EnvVar("SNOWFLAKE_PURINA_USER"),
      password=EnvVar("SNOWFLAKE_PURINA_PASSWORD"),
    ),
 }
)

In this example, the snowflake_resource_key is a SnowflakeResource that has access to the query_history table.

Once the pipeline runs, Snowflake credits will be visible in the Insights tab in the Dagster UI: