Conditionally Run DbtAsset Materialization from an Asset

[rover-wagnar]

Hello!

I'm working on a data pipeline that needs to run monthly. It also needs to be resilient so in the case of failures or cancellations it can be re-run without reprocessing a large amount of data. To accomplish this, I'm using an S3 state file to track the processing of the data as it gets processed in batches.

At the beginning of this pipeline I want to run a DBT snapshot from a dbt_asset, but ONLY the first time the job runs (so if it was killed or stopped and re-run, it would not run the snapshot again). I tried following the advice around yielding MaterializeResult in discussion 19632, but this did not seem to work. This seemed to run but not actually trigger the snapshot to run.

What I would like ideally is to have the pipeline trigger a DBT snapshot run and then wait for the snapshot to finish. Is there a way to conditionally run a dbt_asset from a normal asset and wait for the dbt_asset process to complete?

Here's some of the code for the pipeline:

dbt_asset - correctly selects the dbt snapshot

@dbt_assets(
    manifest=dbt_manifest_path,
    dagster_dbt_translator=DagsterDbtTranslator(
        settings=DagsterDbtTranslatorSettings(enable_asset_checks=True)
    ),
    select="config.meta.dagster.group:data_snapshot",
)
def data_snapshot(
    context: AssetExecutionContext, dbt: DbtCliResource, config: DbtConfig
):
    dbt_build_args = ["build"]
    if config.debug_logs:
        dbt_build_args = ["--log-level", "debug", "build"]
    dbt_run_invocation = dbt.cli(dbt_build_args, context=context)
    yield from dbt_run_invocation.stream()

multi_asset - conditionally runs the dbt_asset

@multi_asset(
    group_name="data_export",
    compute_kind="python",
    specs=[AssetSpec("data_snapshot", skippable=True)]
)
def conditionally_run_snapshot(
    context: AssetExecutionContext,
    config: ExportConfig
):
    """Conditionally runs the data dbt snapshot.
        https://github.com/dagster-io/dagster/discussions/19632
    """
    runtime_start_dt: datetime = datetime.utcnow()
    
    state: ExportStateFile = ExportStateFile.load_state_from_s3(
        s3_path_prefix=config.s3_path_prefix,
        bucket_name=config.bucket_name,
    )
    logger.info(f"snapshot_taken = {str(state.snapshot_taken)}")
    
    if not state.snapshot_taken:
        logger.info("Running data_snapshot ...")
        yield MaterializeResult(asset_key="data_snapshot")
        
        logger.info("data_snapshot completed, updating ExportStateFile ...")
        
        state.snapshot_taken = True
        state.last_job_runtime = runtime_start_dt
        # state.write_state_to_s3()
        
    return None