https://dagster.io/ logo
Powered by
#ask-community
Title
# ask-community
y

Yevhen Samoilenko

08/16/2022, 10:51 AM
Hi. Our use case - we have an assets based job with lots of integrations assets -> dbt assets (created using load_assets_from_dbt_manifest) -> dbt-dependent assets. If some of the integrations assets fails - dbt and downstream assets do not execute at all. Is there any way to make dbt assets run for successful integrations? And only after that fail the job itself.
s

sandy

08/16/2022, 6:49 PM
@owen - thoughts on this?
o

owen

08/16/2022, 11:28 PM
hmm -- I don't think there's a great way of making this work at the moment, just because Dagster's execution model doesn't support these sorts of "partial failure/execution" situations at the moment (even for regular ops / jobs). I definitely see the value here though. We're thinking about more flexible / "smarter" execution options, so this is something to keep in mind. Are your integration assets created using a multi-asset (i.e. a single step that creates a bunch of assets), or is each integration asset its own distinct step?
y

Yevhen Samoilenko

08/17/2022, 12:40 PM
Are your integration assets created using a multi-asset (i.e. a single step that creates a bunch of assets), or is each integration asset its own distinct step?
Actually, we use both of these approaches.
Maybe we could use this new output_required feature for this? And use AssetObservation (or something else) for warnings. Something like
Copy code
from dagster import asset, Output, AssetObservation, OpExecutionContext


@asset(output_required=False)
def may_fail(context: OpExecutionContext):
    try:
        result = [1, 2, 3, 4]

        yield Output(result)
    except Exception as e:
        yield AssetObservation(
            asset_key=context.asset_key_for_output(),
            metadata={"error": str(e)},
        )


@asset
def downstream(may_fail):
    return may_fail + [5]
In this case, the job won't fail. And theoretically, downstream assets should execute for successful materializations. The only thing left is to somehow check for warnings and fail the job at the end (or send an alert to slack or something like that). @sandy, @owen what do you think?
o

owen

08/17/2022, 4:18 PM
In theory, what you're describing should likely work. but in practice (due to the specifics of how we actually execute these dbt assets), I don't believe this would work. Essentially, even if 
output_required
is set to False, the steps downstream of the optional output will not be executed, not just the assets. Because all dbt assets run in a single step, they will all be skipped. This is pretty confusing behavior, and should be fixed (but I don't think it's a super quick fix on our end to do that). Just created a ticket for that here: https://github.com/dagster-io/dagster/issues/9408
👍 1
y

Yevhen Samoilenko

08/18/2022, 7:54 AM
And what about this approach:
Copy code
from dagster import (
    define_asset_job,
    AssetSelection,
    run_status_sensor,
    DagsterRunStatus,
    RunRequest,
    DefaultSensorStatus,
)


integrations_job = define_asset_job(
    name="integrations_job",
    selection=AssetSelection.groups("integrations"),
)
dbt_job = define_asset_job(
    name="dbt_job",
    selection=AssetSelection.groups("dbt"),
)


@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    request_job=dbt_job,
    monitored_jobs=[integrations_job],
    default_status=DefaultSensorStatus.RUNNING,
)
def integrations_job_success_sensor():
    return RunRequest(run_key=None)


@run_status_sensor(
    run_status=DagsterRunStatus.FAILURE,
    request_job=dbt_job,
    monitored_jobs=[integrations_job],
    default_status=DefaultSensorStatus.RUNNING,
)
def integrations_job_failure_sensor():
    return RunRequest(run_key=None)
What if we split our asset-based jobs into two parts - integrations and dbt? And use run_status_sensor for triggering dbt run for both failure and success statuses? I know, this is kind of a hack, but still. And is it possible to run dbt assets only for successful integrations (using run_config or something else)?
o

owen

08/18/2022, 4:30 PM
Currently, this is not possible, but we're considering extending RunRequest to have an asset_selection argument (which is what you'd want here I believe)
y

Yevhen Samoilenko

08/18/2022, 4:37 PM
Yep, this is exactly what I want! Thanks. I think we'll stick with this approach for some time until we come up with some more elegant solution 🙂
2 Views
Powered by