Exporting metrics from Dagster+ Insights#

Using a GraphQL API endpoint, you can export Dagster+ Insights metrics from your Dagster+ instance.

Refer to the Available Insights metrics for a list of available metrics.

Prerequisites#

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

• A Dagster+ account
• Access to the Dagster+ Insights feature
• A Dagster+ user token
• Your deployment-scoped Dagster+ deployment URL. For example: dagster-university.dagster.cloud/prod

Before you start#

Before you start, note that:

• Metrics are currently computed once per day
• We don't recommend frequently querying over large time ranges that may download a large amount of data. After an initial data load, we recommend loading data daily for the most recent week or less.

Using the API#

In this example, we're using the GraphQL Python Client to export the Dagster credits metric for all assets for September 2023:

from datetime import datetime
from dagster_graphql import DagsterGraphQLClient

ASSET_METRICS_QUERY = """
query AssetMetrics($metricName: String, $after: Float, $before: Float) {
  reportingMetricsByAsset(
    metricsSelector: {
      metricName: $metricName
      after: $after
      before: $before
      sortAggregationFunction: SUM
      granularity: DAILY
    }
  ) {
    __typename
    ... on ReportingMetrics {
      metrics {
        values
        entity {
          ... on ReportingAsset {
            assetKey {
              path
            }
          }
        }
      }
    }
  }
}

"""


def get_client():
    url = "YOUR_ORG.dagster.cloud/prod"  # Your deployment-scoped url
    user_token = "YOUR_TOKEN"  # A token generated from Organization Settings > Tokens
    return DagsterGraphQLClient(url, headers={"Dagster-Cloud-Api-Token": user_token})


if __name__ == "__main__":
    client = get_client()
    result = client._execute(
        ASSET_METRICS_QUERY,
        {
            "metricName": "__dagster_dagster_credits",
            "after": datetime(2023, 9, 1).timestamp(),
            "before": datetime(2023, 10, 1).timestamp(),
        },
    )

    for asset_series in result["reportingMetricsByAsset"]["metrics"]:
        print("Asset key:", asset_series["entity"]["assetKey"]["path"])
        print("Daily values:", asset_series["values"])

To use this example yourself, replace the values of url and user_token in this function:

def get_client():
    url = "YOUR_ORG.dagster.cloud/prod"  # Your deployment-scoped url
    user_token = "YOUR_TOKEN"  # A token generated from Organization Settings > Tokens
    return DagsterGraphQLClient(url, headers={"Dagster-Cloud-Api-Token": user_token})

Refer to the Reference section for more info about the endpoints available in the GraphQL API.

Reference#

For the full GraphQL API reference:

1. Navigate to https://YOUR_ORG.dagster.cloud/prod/graphql, replacing YOUR_ORG with your organization name. For example: https://dagster-university.dagster.cloud/prod/graphql
2. Click the Schema tab.

Available top-level queries#

reportingMetricsByJob(
  metricsFilter: JobReportingMetricsFilter
  metricsSelector: ReportingMetricsSelector!
): ReportingMetricsOrError!

reportingMetricsByAsset(
  metricsFilter: AssetReportingMetricsFilter
  metricsSelector: ReportingMetricsSelector!
): ReportingMetricsOrError!

reportingMetricsByAssetGroup(
  metricsFilter: AssetGroupReportingMetricsFilter
  metricsSelector: ReportingMetricsSelector!
): ReportingMetricsOrError!

Specifying metrics and time granularity#

Use metricsSelector to specify the metric name and time granularity:

input ReportingMetricsSelector {
  after: Float # timestamp
  before: Float # timestamp
  metricName: String # see below for valid values
  granularity: ReportingMetricsGranularity
}

enum ReportingMetricsGranularity {
  DAILY
  WEEKLY
  MONTHLY
}

# The valid metric names are:
# "__dagster_dagster_credits"
# "__dagster_execution_time_ms"
# "__dagster_materializations"
# "__dagster_step_failures"
# "__dagster_step_retries"
# "__dagster_asset_check_errors"
# "__dagster_asset_check_warnings"