Create a custom freshness check which operates across all partitions

[dpeng817]

We don't currently have full partitions support for asset checks. In the meantime, you can construct an asset check which checks all partitions of a given asset. Here's a code example to demonstrate how to do this for freshness specifically.
When constructed in this way, the freshness check can be passed to the build_freshness_check_sensor API, which will automatically run the custom freshness check for you.

from datetime import timedelta
from typing import Sequence

import pendulum
from dagster import asset_check
from dagster._core.definitions.asset_check_factories.freshness_checks.sensor import (
    build_sensor_for_freshness_checks,
)
from dagster._core.definitions.asset_check_factories.utils import FRESH_UNTIL_METADATA_KEY
from dagster._core.definitions.asset_check_result import AssetCheckResult
from dagster._core.definitions.asset_check_spec import AssetCheckSeverity
from dagster._core.definitions.metadata.metadata_value import TimestampMetadataValue
from dagster._core.execution.context.compute import AssetCheckExecutionContext
from dagster_snowflake.resources import SnowflakeResource

EXPECTED_LAST_UPDATED_RECORD_TIME_DELTA = timedelta(days=1)


@asset_check(
    description="Check freshness across all partitions",
    asset=<your_asset_here>,
)
def custom_freshness_check_all_partitions(
    context: AssetCheckExecutionContext, ...
) -> AssetCheckResult:
    current_timestamp = pendulum.now("UTC").timestamp()
    # This demonstrates how to get all dynamic partitions specifically, and might need to be altered depending on your 
    # specific kind of partitions.
    partitions = YOUR_PARTITIONS_DEF.get_partition_keys(
        dynamic_partitions_store=context.instance
    )
    failed_partitions_info: Sequence[Dict[str, str]] = [] # Store info about each failure.
    for partition in partitions:
        last_updated_time = ... # your code here for retrieving last update time for each partition.
        if (
            current_timestamp - last_updated
            > EXPECTED_LAST_UPDATED_RECORD_TIME_DELTA.total_seconds()
        ):
            failed_partitions_info.append(
                {
                    "partition": partition,
                    "additional_info": ...,
                    "error_descriptions": [
                        f"Last updated time is stale by more than 1 day for partition {partition}"
                    ],
                }
            )

    if failed_partitions:
        return AssetCheckResult(
            passed=False,
            severity=AssetCheckSeverity.WARN,
            metadata={
                failed_partitions_info: failed_partitions_info,
            },
            description=f"{len(failed_partitions_info)} partitions are not fresh.",
        )

    return AssetCheckResult(
        passed=True,
        severity=AssetCheckSeverity.WARN,
        description="All partitions are fresh.",
        metadata={
            # This allows the sensor to know when to run the freshness check next.
            FRESH_UNTIL_METADATA_KEY: TimestampMetadataValue(
                current_timestamp + EXPECTED_LAST_UPDATED_RECORD_TIME_DELTA.total_seconds()
            )
        },
    )


freshness_sensor = build_sensor_for_freshness_checks(
    freshness_checks=[custom_freshness_check_all_partitions],
    minimum_interval_seconds=300,
    name="dynamic_partitions_freshness_sensor",
)

Note that if you make a blocking asset check in this style, it will block all downstream partitions, not just the ones that failed.

[dpeng817]

answer in description

[loerinczy]

Thank you very much for the detailed description!

I am trying to implement such a freshness check logic with a sensor on the freshness checks (via the provided build_sensor_for_freshness_checks function), however, the sensor does not find the asset that the asset check is attached to in the Definitions object. All of the assets have dynamic partitions.

This is the issue I created for the bug: ISSUE

In order to reproduce the situation, here is a minimal example with dagster 1.7.1, where the code location contains the following code alone in the _init_.py file:

from dagster import (
    Definitions,
    asset,
    asset_check,
    AssetCheckResult,
    DynamicPartitionsDefinition,
    AssetCheckSeverity,
    build_sensor_for_freshness_checks,
)

@asset(
        partitions_def=DynamicPartitionsDefinition(name="b"),
)
def first_asset():
    return 1


@asset(
        partitions_def=DynamicPartitionsDefinition(name="a")
)
def second_asset():
    return 1


@asset_check(
    asset=first_asset,
    metadata={"dagster/freshness_params": {"dagster/lower_bound_delta": 10}},
)
def first_asset_check():
    return AssetCheckResult(
        passed=False,
        severity=AssetCheckSeverity.WARN,
        description="something",
        metadata={"reason": "something"},
    )


freshness_sensor = build_sensor_for_freshness_checks(
    freshness_checks=[first_asset_check],
    minimum_interval_seconds=30,
    name="freshness_check_sensor",
)


defs = Definitions(
    assets=[first_asset, second_asset], asset_checks=[first_asset_check], sensors=[freshness_sensor]
)

My experiments suggest, that the issue occurs when the second asset has a DynamicParititonsDefinition with a name that precedes that of the first asset in the alphabet. If you swap the names in the example below, the error disappears. This is just an empirical finding, to hint at a direction.

Please take a look at this issue and propose a way how to proceed.
Thank you very much for your assistance in advance! :)

[gibsondan]

Commented on the linked issue, but we have a fix for this bug coming out this week in the dagster 1.8.1 release.