I'm doing some live testing during the presentation 🙂 I try to load all my Dags as assets:

dag_bag = DagBag(
        dag_folder=dag_path,
        include_examples=False,  # Exclude Airflow example dags
        safe_mode=safe_mode,
    )
    assets = []
    for dag in list(dag_bag.dags.values()):
        assets = assets + load_assets_from_airflow_dag(dag)


    return Definitions(
        assets=assets,
        executor=executor,
        loggers=loggers,
    )

It seems that this is a bad idea because I have this mistake:

dagster._core.errors.DagsterInvalidDefinitionError: Conflicting versions of resource with key 'airflow_db' were provided to different assets. When constructing a job, all resource definitions provided to assets must match by reference equality for a given key.

But I don't get why 😕

@Alexandre Guitton thanks for the report, unfortunately multiple`load_assets_from_airflow_dag` calls in a repo/definitions object creates conflicting resource defs.

I also tested it, and I encountered some drawbacks: • If an operator uses a pool, it "fails" (even if that step succeed) because the pool is not defined (and since retry is not implemented it stop at first try).

[2023-02-08 18:11:00,419] {taskinstance.py:1025} INFO - Dependencies not met for <TaskInstance: wkf_github_api_rate.collect manual__2023-02-08T18:09:44.596652+00:00 [None]>, dependency 'Pool Slots Available' FAILED: ("Tasks using non-existent pool '%s' will not be scheduled", 'process_in_redshift')
[2023-02-08 18:11:00,424] {taskinstance.py:1226} WARNING - 
--------------------------------------------------------------------------------
[2023-02-08 18:11:00,424] {taskinstance.py:1227} WARNING - Rescheduling due to concurrency limits reached at task runtime. Attempt 1 of 7. State set to NONE.

• If my job fails at a step, I can't restart over from the step that failed, because:

[2023-02-08 18:12:31,901] {taskinstance.py:1025} INFO - Dependencies not met for <TaskInstance: wkf_github_api_rate.collect manual__2023-02-08T18:09:44.596652+00:00 [None]>, dependency 'Trigger Rule' FAILED: Task's trigger rule 'all_success' requires all upstream tasks to have succeeded, but found 1 non-success(es). upstream_tasks_state={'total': 1, 'successes': 0, 'skipped': 0, 'failed': 0, 'upstream_failed': 0, 'done': 0}, upstream_task_ids={'is_hourly'}

That's why I wanted to check the behavior of the asset in these cases.

the asset apis will still have those same issues atm

And there's not equivalent of clear tasks keeping upstream succeed either?

in dagster there isn't really a concept of clearing tasks but our retry system accomplishes a similiar thing https://docs.dagster.io/guides/dagster/re-execution but like i mentioned until dagster-airflow finds a way to persist its airflow db state it doesn't work (you'll need to rerun the full job if a failure occurs)

Super clear 🙂 Thanks!

@Alexandre Guitton 👋 1.2.0 shipped with a way to persist airflow db state https://dagster.slack.com/archives/CH2SCAV19/p1678471767802909

I saw 🎉 I plan to test it this week 😎