Hello, is it possible to implement the following DAG with dagster such that solids A,B,C,D all operate together for each file in a directory using dynamic outputs? I couldn't seem to find similar examples in previous slack messages or in the docs. Any responses would be greatly appreciated! Thanks

That should be possible but it might be a bit confusing depending on the inputs expected to Solid A-C and Solid D.

files = get_files() # retrieve all files as a list
a_result = files.map(solid_a).collect() # call solid_a method for all files and collect result as a list
b_result = files.map(solid_b).collect()
c_result = files.map(solid_c).collect()
solid_d(a=a_result, b=b_result, c=c_result)

Thanks Anton. Ideally I would like to call solid_d for each file result just to keep the logic a bit simpler in the solid. Is there a way to do this?

I think so but unfortunately you cannot have nested DynamicOutputs currently. I find the current implementation rather limiting and difficult to use but it is an experimental feature that the Dagster team is working on. I have expressed some of my concerns to the team here https://github.com/dagster-io/dagster/issues/4200 Here is a rough idea of how I think it would work.

files = get_files() # retrieve all files as a list
a = files.map(solid_a)
b = files.map(solid_b)
c = files.map(solid_c)
a.map(solid_d)
b.map(solid_d)
c.map(solid_d)

Thanks Anton for the reply! The previous solution you mentioned works well as a compromise but I agree it would be nice for some slightly better options for this type of pipeline in the future.