kubeflow condition and parallel tasks

We are going to look at a simple kubeflow condition and parallel task when working with pipeline, This pipeline will greet pet owner and if there's pets we greet them too. 

This is our example components:-

@component

def sayhi(name: str) -> int:

  return 1 if name == "eric" else 2

@component

def args_generator_op() -> list:

    return [{"cats": "1", "dogs": "2"}, {"cats": "10", "dogs": "20"}]

And this is our pipeline code:-

import json

from kfp import dsl

@dsl.pipeline(name="control", pipeline_root=PIPELINE_ROOT)

def pipeline():

    name = sayhi(name="greet-human")

 

    with dsl.If(name.output == 1, "if has pets"):    

      cat_and_dog = args_generator_op()

    

    with dsl.Else("if has pets"):

      cat_and_dog = args_generator_op()

      with dsl.ParallelFor(cat_and_dog.output, name="hi-to-pets") as item:

          print_op(msg=json_string)

   

Let's run our pipeline

compiler.Compiler().compile(

    pipeline_func=pipeline, package_path="control_pipeline.yaml"

)

     

And then execute it

DISPLAY_NAME = "control"

job = aiplatform.PipelineJob(

    display_name=DISPLAY_NAME,

    template_path="control_pipeline.yaml",

    pipeline_root=PIPELINE_ROOT,

)

job.run()

What our pipeline would look like:-

And if we run our pipeline, we can see it perform if checks. If results == 0 it

won't run other tasks and this is what it looks like:-

You can see the results too and what is the results given or returned

   

And if we pass a value called "eric" then it would run the parallel task and

this is what it looks like. You can clearly see that the parallel task runs.