Assets do not show up as consumed

[bolkedebruin]

Apache Airflow version

3.0.1

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I am playing around with Assets.

This is the generating DAG / Asset (Confusing!)

from airflow.sdk import Metadata, asset

@asset(uri="file:///tmp/example.csv", schedule=None)
def write_example(self):
    with open("/tmp/example.csv", "w") as f:
        f.write("Hallo")

    yield Metadata(self, {"row_count": 0})

I have two consuming DAGs, one "asset" syntax based and one taskflow based:

from airflow.sdk import Asset, asset
from airflow.sdk import DAG, dag, task

write_example = Asset("write_example")

#@asset(schedule=None)

@dag
def my_dag():

    @task
    def consume_csv(write_example):
        print("Bla")

my_dag()

and

from airflow.sdk import Asset, asset
from airflow.sdk import DAG, dag, task

@asset(schedule=None)
def consume_csv(write_example):
    print("Bla")

Neither of the two DAGs generate a "consume" number at the write_example asset / dag

What you think should happen instead?

The number of consuming DAGs should read "2" for the "write_example" asset

How to reproduce

see above

Operating System

Linux

Versions of Apache Airflow Providers

not relevant

Deployment

Docker-Compose

Deployment details

not relevant

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[uranusjr]

Currently assets are only shown as consumed if they are used in schedule, not inlets. I feel we might want to do some more thinking instead of just adding those. It’s probably best to have three fields: producing (things that list the asset as outlet), consuming (list as inlet), and scheduled (list in schedule). Not sure about the names here.

cc @bbovenzi for ideas.

[bolkedebruin]

I can understand that, however the examples I shared above are inlets, i.e. consumed by the task.

[uranusjr]

Yes I noticed. I think the term consuming is confusing anyway.

[bolkedebruin]

Ah gotcha, I misread your comment. It is now confusing indeed. To me "consumed" means "used worked on". We could consider the following alternatives:

Data Dependency / Triggers (or Schedules to avoid naming conflicts)

or

Consumed by / Triggers (or Schedules)

[bolkedebruin]

Adding:

from airflow.sdk import Asset, asset
from airflow.sdk import DAG, dag, task

write_example = Asset("write_example")

#@asset(schedule=None)

@dag
def my_dag():

    @task
    def consume_csv(write_example):
        print("Bla")

my_dag()

This sets it as a schedule and a it is an inlet. However

1. When triggering the Asset (write_example) this dag does not get kicked off
2. It also does not become visible as "consumed" in the Asset

I might be doing something unexpected write_example = Asset("write_example"), but how to do it otherwise?