CODEWITHSUNDEEP
pythonkubeflowkubeflow-pipelines
user3599803
user3599803

Reputation: 7044

kubeflow pipeline dynamic output list as input parameter

I use a ParallelFor over a dynamic list. I want to collect all the outputs from the loop, and pass them to another ContainerOp.
Something like the following, which obviously does not work, since the outputs list is will be static.

with dsl.ParallelFor(op1.output) as item:
    op2 = dsl.ContainerOp(
      name='op2',
      ...
      file_outputs={
         'outputs': '/outputs.json',
    })
    outputs.append(op2.output)


op3 = dsl.ContainerOp(
   name='op3',
   ...
   arguments=['--input': outputs]  # won't work
)

Upvotes: 7

Views: 5272

Answers (3)

Ezhik
Ezhik

Reputation: 874

Unfortunately, the solution of Ark-kun is not working for me. But there is a simple way to implement fan-in workflow if we know the number of inputs in advance. We may precalculate pipeline DAG like that:

@kfp.components.create_component_from_func
def my_transformer_op(item: str) -> str:
    return item + "_NEW"


@kfp.components.create_component_from_func
def my_aggregator_op(items: list) -> str:
    return "HELLO"


def pipeline(array_of_arguments):
    @dsl.pipeline(PIPELINE_NAME, PIPELINE_DESCRIPTION)
    def dynamic_pipeline():
        outputs = []
        for i in array_of_arguments:
            outputs.append(my_transformer_op(str(i)).output)
        my_aggregator_op(outputs)
    return dynamic_pipeline

...

    run_id = client.create_run_from_pipeline_func(
        pipeline(data_samples_chunks), {},
        run_name=PIPELINE_RUN,
        experiment_name=PIPELINE_EXPERIMENT).run_id

Pipeline graph

Upvotes: 5

santiago92
santiago92

Reputation: 423

I have run into issues with dynamic "fanning-out" and then "fanning-in" with Kubeflow Pipelines as well. Maybe a little heavy-handed but I used a mounted PVC claim to get over this.

Kubeflow allows you to mount a known PVC or create a new one on the fly using VolumeOp (link here). This snippet shows how to use a known PVC.

    pvc_name = '<available-pvc-name>' 
    pvc_volume_name = '<pvc-uuid>' # pass the pvc uuid here

    # Op 1 creates a list to iterate over
    op_1 = dsl.ContainerOp(
            name='echo',
            image='library/bash:4.4.23',
            command=['sh', '-c'],
            arguments=['echo "[1,2,3]"> /tmp/output.txt'],
            file_outputs={'output': '/tmp/output.txt'})

    # Using withParam here to iterate over the results from op1
    # and writing the results of each step to its own PVC
    with dsl.ParallelFor(op_1.output) as item:
        op_2 = dsl.ContainerOp(
            name='iterate',
            image='library/bash:4.4.23',
            command=['sh', '-c'],
            arguments=[f"echo item-{item} > /tmp/output.txt; "  # <- write to output  
                       f"mkdir -p /mnt/{{workflow.uid}}; "  # <- make a dir under /mnt
                       f"echo item-{item}\n >> /mnt/{{workflow.uid}}"],  # <- append results from each step to the PVC
            file_outputs={'output': '/tmp/output.txt'},
            # mount the PVC
            pvolumes={"/mnt": dsl.PipelineVolume(pvc=pvc_name, name=pvc_volume_name)})

    op_3 = dsl.ContainerOp(
            name='echo',
            image='library/bash:4.4.23',
            command=['sh', '-c'],
            arguments=[f"echo /mnt/{{workflow.uid}} > /tmp/output.txt"],
            # mount the PVC again to use
            pvolumes={"/mnt": dsl.PipelineVolume(pvc=pvc_name, name=pvc_volume_name)},
            file_outputs={'output': '/tmp/output_2.txt'}).after(op_2)

Ensure that op_3 runs after the loops from op_2 using after(op_2) in the end.

Note: This might be a heavy-handed approach and there might be better solutions if KFP allows this as part of the KF compiler but I couldn't get it to work. If it's easy to create a PVC in the env this might work for your case.

Upvotes: 4

Ekaba Bisong
Ekaba Bisong

Reputation: 2982

The issue is that op3 is not properly referencing the output from op2 as an input argument. Try this:

op3 = dsl.ContainerOp(
    ...
    arguments=['--input': op2.outputs['outputs']]
)

Upvotes: 0

Related Questions