Asset checks: Try them out

[johannkm]

Important

Dagster asset checks are currently an experimental feature, and the API may change in future releases. During this period, checks do not consume Dagster Cloud Credits.

---

We've recently released a new @asset_check API as described in RFC: Asset checks. It includes:

• A new @asset_check decorator for defining checks that run in their own Op
• A new check_specs argument on @asset for defining checks that execute in the same Op as the materialization
• An integration with DBT Tests: [RFC] Integrating dbt tests with asset checks #16527
• UI:

We'd love to get your feedback after trying these out! A basic example to get started:

from dagster import asset, asset_check, AssetCheckResult, Definitions

@asset
def my_asset():
  pass

@asset_check(asset=my_asset, description="A very useful check")
def always_pass_check():
  return AssetCheckResult(passed=True)

defs = Definitions(assets=[my_asset], asset_checks=[always_pass_check])

This check will run whenever you materialize my_asset.

See the docs for more.

---

Share feedback here, or in the #data-quality-asset-checks Slack channel.

[abi-mutinex]

Great work here team! A quick question - I know there were plans to integrate dbt tests to this. Any idea when they would come to life?

[johannkm]

We plan to ship dbt support in the release this Thursday

[rexledesma]

Hi @abi-mutinex, they are now released in dagster-dbt==0.20.13. Check out the discussion for more details: #16527.

[dduong1603]

A couple of feedback/observations when trying out the asset checks:

• Not sure how to feel about the current UI color when the asset is materialized successfully but some checks fail 😅 I think I would still prefer the Materialized status to be green because at a quick glance I thought the asset failed to materialize.
• Similarly, I feel like there's a need for additional run statuses to distinguish an actual failed run that fails both materialization and checks, and a run that just fails some checks. Right now they all get grouped into Failure status.
• I still prefer the check bubble to be separated in one of the mocked UI. It looks cleaner for me 😅
• I get this error when running some dbt tests (the test passed...I guess it just returned something that the check metadata didn't expect):

dagster._core.errors.DagsterInvariantViolationError: Received unexpected AssetCheckResult. It targets asset 'elementary/filtered_information_schema_columns' which is not targeted by any of the checks currently

Stack Trace:                                                                                                                                                                                                       
   File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_plan.py", line 273, in dagster_event_sequence_for_step                                                                        
     for step_event in check.generator(step_events):                                                                                                                                                                
   File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 475, in core_dagster_event_sequence_for_step                                                                   
     for user_event in _step_output_error_checked_user_event_sequence(                                                                                                                                              
   File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 158, in _step_output_error_checked_user_event_sequence                                                         
     for user_event in user_event_sequence:                                                                                                                                                                         
   File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 94, in _process_asset_results_to_events                                                                        
     yield from _process_user_event(step_context, user_event)                                                                                                                                                       
   File "/usr/local/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 129, in _process_user_event                                                                                    
     asset_check_evaluation = user_event.to_asset_check_evaluation(step_context)                                                                                                                                    
                              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                                                                    
   File "/usr/local/lib/python3.11/site-packages/dagster/_core/definitions/asset_check_result.py", line 88, in to_asset_check_evaluation                                                                            
     raise DagsterInvariantViolationError(
{"data": {"execution_time": 2.1847, "index": 6, "name": "elementary_schema_changes_dim__dates_", "node_info": {"materialized": "test", "meta": {"dagster": {"asset_check": true}}, "node_finished_at": "2023-10-02 
 Elementary: Started running data monitors on: WAREHOUSE.ANALYTICS_DUKE.DIM__DATES quarter_end                                                                                                                      
 Exception ignored in: <generator object DbtCliInvocation.stream_raw_events at 0x7fd4b0410f40>                                                                                                                      
 Traceback (most recent call last):                                                                                                                                                                                 
   File "/usr/local/lib/python3.11/site-packages/dagster_dbt/core/resources_v2.py", line 316, in stream                                                                                                             
     yield from event.to_default_asset_events(                                                                                                                                                                      
 RuntimeError: generator ignored GeneratorExit

[dduong1603]

@rexledesma hmm I set DBT_INDIRECT_SELECTION=cautious in my user code deployment but still see the same issue. I didn't select any parent model though, and when running locally everything was ok. The failure can only be seen on the k8s deployment

[rexledesma]

@dduong1603 Glad it's running locally now. The environment variable might not be getting threaded properly in your deployment. Try setting it directly in code os.environ["DBT_INDIRECT_SELECTION"] = "cautious" or wait for dagster-dbt==0.21.1 where we we'll land #16801 and you won't have to configure this.

[dduong1603]

Ah nice setting the variable directly in code does work!

[josh-acuitymd]

Hi @rexledesma !! Is there a way to override the default behavior of setting DBT_INDIRECT_SELECTION=empty? Based on this it looks like it might ignore any assignments I make whether that's eager, cautious, or whatever.

[rexledesma]

See #18891 (comment) for an explanation of this behavior. When you enable asset checks, we require DBT_INDIRECT_SELECTION=empty since Dagster takes control of the exact dbt tests that are run in an execution.

[dridk]

Hi,

Thank you for this cool feature.

I recently tried mage.ai, which I find more intuitive ( more python way) than Dagster (POO java way) in several aspects.
For data validation, for example, mage.ai uses assertion like any other python test.

https://docs.mage.ai/development/data-validation

@test 
def test_number_of_rows(df) -> None:         
assert len(df.index) >= 2, 'The output has more than 1 row.'

assert df.isnull().sum() == 0 , "null values"

With dagster, you have to make one function per test. Or use the yield operator in the asset.

Why not use something like this:

@asset_check 
def test1(df):
      assert len(df) > 0
      assert df.isnull().sum() = 0 

@asset_check
def test2(df): 
    ....

[johannkm]

Thanks for the feedback! The concise syntax is nice.

We have a couple concerns about using assert:

• asserts can be disabled at runtime with python -O or envvar PYTHONOPTIMIZE
• AssetCheckResult can be used inside @asset_check, or inside an @asset with multiple checks so you need to disambiguate which check you're evaluating. assert would only work with the former
• what if a 3rd party library you invoke calls assert? Should that get the overloaded behavior?

For your test1 example, how would you feel about breaking it in to separate checks so the UI can display which check failed?

@asset_check(asset=df)
def check_not_empty(df):
  return AssetCheckResult(passed=len(df)==0)

@asset_check(asset=df)
def check_no_nulls(df):
  num_nulls = df.isnull().sum()
  return AssetCheckResult(passed=num_nulls>0, metadata={"number of nulls": num_nulls)

There's certainly more we can do to make the syntax friendly, so we're happy to get more suggestions

[dridk]

Ok ! It makes sense , particulary the point 3 ! I think it is fine in this way

[rob-apella]

can an asset check depend on multiple assets? e.g., in DBT, a singular test may join together two DBT models to perform a data test. this test would need to run after both models have been materialized.

edit: found this #16725

[rexledesma]

We currently only support dbt generic tests as Dagster asset checks.

We're tracking support for singular tests (and generally, relationship tests that depend on multiple models) here: #17628

[rob-apella]

awesome thanks @rexledesma

[johannkm]

more specific tracking issue: #18440

[sisidra]

We want to define asset checks on source/external assets. That seems to be partially supported.
Issues:

• Dagster Web buttons "Execute" and "Execute All" are disabled
• Creating a dedicated job per check is not reasonable for large amount of assets * checks.
• External asset lack IO Manager properties (although the docs state that the goal is to replace Source Assets with External Assets) - this is more related to external assets as it is hard to integrate external assets into asset graph.

[johannkm]

Would you mind filing an issue for Dagster Web buttons "Execute" and "Execute All" are disabled?

[smackesey]

External asset lack IO Manager properties (although the docs state that the goal is to replace Source Assets with External Assets) - this is more related to external assets as it is hard to integrate external assets into asset graph.

External asset APIs are under heavy development right now, we will definitely support IO managers on them.