Hi all, I'm new to dagster and I'm experimenting with it for my application. I have some periodic refresh job I need to run, and I need to refresh multiple items, and need the graph to be built at runtime. I basically need to spawn multiple sub-jobs from a loop -- this way I'd like to be able to tell if a particular refresh job failed. So something like this, but it doesn't work:

@asset
def refresh_item(item):
    ... some code

@asset
def refresh_catalogue():
    catalogue: list[MyItem] = load_catalogue()

    for item in catalogue:
        refresh_item(item)

I've not been able to find any usable examples of this online -- the one I found uses

pipelines

which appears to be deprecated. How would I do this in dagster?

Graph assets + dynamic graphs OR partitions, depending on whether the list you're fanning out over grows and shrinks or not

Thanks for the links! The dynamic graph link worked like a charm 😄, I followed the example on the page but I'm not sure how I should include

graph-asset

or

graph-multi-asset

. I tried replacing

@job

with them but the job wouldn't run. This is what I ended up with:

@op
def refresh_item(item):
    ... some code

@op(out=DynamicOut())
def span_jobs():
    catalogue = load_catalogue()

    for i, specs in enumerate(catalogue):
        yield DynamicOutput(specs, mapping_key=str(i))

@job
def refresh_catalogue():
    jobs = span_jobs()
    _ = jobs.map(refresh_item)

If you want it to be an asset, then you can use @graph_asset, then either just materialize the asset, or make it into an asset job, which will run similarly to an op-based job

Sorry, what do you mean by materialize the asset?

Is your graph asset just

@graph_asset
def refresh_catalogue():
    jobs = span_jobs()
    _ = jobs.map(refresh_item)

? What are the errors you're getting

Error I'm getting, where I return

jobs.map(refresh_item).collect()

dagster._core.errors.DagsterInvalidDefinitionError: @graph 'refresh_catalogue' returned problematic value of type <class 'dagster._core.definitions.composition.DynamicFanIn'>. Expected return value from invoked node or dict mapping output name to return values from invoked nodes

Ah yeah that's one oddity about graph assets, because it's an asset you have to return something from the graph

Ok I got it to work from reading https://github.com/dagster-io/dagster/issues/15664, but it's like 🤦‍♂️. I have to create a dummy function with

@op

and return an empty dataframe:

@op
def collector(stub):
    return pd.DataFrame()

@graph_asset
def refresh_catalogue():
    jobs = span_jobs()
    stub = jobs.map(refresh_item).collect()
    return collector(stub)

Maybe a bit pedantic but it doesn't really have to be a fake dataframe, you could return an empty dict or something like that too. But yeah it's kinda weird having to return something if you didn't really plan ahead to rely on IO managers

Makes sense. Anyway, thanks for your help