How do I collect outputs of components in an inner loop and run a component before continuing with the next iteration of the outer loop?

[cy-rae]

Hi there!

I do not understand how I can collect the outputs from an inner loop and run a component that uses the outputs as a parameter before the next outer loop iteration starts. Here is an example:

def my_pipeline():
    dsl.ParallelFor(["a", "b", "c"], parallelism=1) as x:
          dsl.ParallelFor([1, 2, 3]) as y:
               foo_task = foo(x=x, y=y)
          foo2(my_list=dsl.Collected(foo_task.output))    # This component has to run before the next iteration starts.

Unfortunately, the next iteration of the outer loop starts before the foo2 component.

I have tried to use nested pipelines but I could not realize the desired behavior. Does dsl.Collected only work "globally" and not "loop-scoped"?

[zazulam]

@cy-rae What version of kfp were you trying this on? We recently just got dsl.collected working and is available in the latest release, 2.14.