<jupyter_start><jupyter_text>Add transformers to 05_01_Pipeline_SDK example and show how to use node selectors to land on preferred nodegroup with label role=workers  Define Tolerations This notebook is tested with Kubeflow v1.1.0 which has multi-tenancy support in Kubeflow pipelines.   Ensure that following labels and taints exists (role=workers) on nodeskubectl label nodes  role=workers      kubectl get nodes --show-labels<jupyter_code>import kfp.dsl as dsl
from kfp import compiler

from kubernetes import client as k8s_client
from kubernetes.client.models import V1EnvVar, V1SecretKeySelector
def node_selector(op):
    if isinstance(op, dsl.ContainerOp):
        op.add_node_selector_constraint('role', 'workers')
        # op.add_node_selector_constraint('single-az', 'true')
        # op.add_node_selector_constraint('spot', 'false')
        # op.container.set_memory_request("2G")
        # op.container.set_cpu_request("1")<jupyter_output><empty_output><jupyter_text>Build simple components and pipelines<jupyter_code>import kfp
authservice_session='authservice_session=<cookie>'
client = kfp.Client(host='http://<aws_alb_host>/pipeline', cookies=authservice_session)
#client.list_experiments(namespace="<your_namespace>")
from kfp import dsl

def add_two_numbers(a, b):
    return dsl.ContainerOp(
        name='calculate_sum',
        image='python:3.6.8',
        command=['python', '-c'],
        arguments=['with open("/tmp/results.txt", "a") as file: file.write(str({} + {}))'.format(a, b)],
        file_outputs={
            'data': '/tmp/results.txt',
        }
    )

def echo_op(text):
    return dsl.ContainerOp(
        name='echo',
        image='library/bash:4.4.23',
        command=['sh', '-c'],
        arguments=['echo "Result: {}"'.format(text)]
    )<jupyter_output><empty_output><jupyter_text>Define your pipeline as a Python function<jupyter_code>@dsl.pipeline(
  name='Calcualte sum pipeline',
  description='Calculate sum of numbers and prints the result.'
)
def calculate_sum(
    a=7,
    b=10,
    c=4,
    d=7
):
    """A four-step pipeline with first two running in parallel."""

    sum1 = add_two_numbers(a, b)
    sum2 = add_two_numbers(c, d)
    sum = add_two_numbers(sum1.output, sum2.output)

    echo_task = echo_op(sum.output)
    
    pipeline_conf = dsl.get_pipeline_conf()
    pipeline_conf.add_op_transformer(node_selector)<jupyter_output><empty_output><jupyter_text>Compile the pipeline<jupyter_code>kfp.compiler.Compiler().compile(calculate_sum, 'calculate-sum-pipeline.zip')<jupyter_output><empty_output><jupyter_text>Deploy pipeline<jupyter_code>#client = kfp.Client()
aws_experiment = client.create_experiment(name='aws', namespace='<your_namespace>')
my_run = client.run_pipeline(aws_experiment.id, 'calculate-sum-pipeline', 
  'calculate-sum-pipeline.zip')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12324438202247191, 'fraction_numerical': 0.009831460674157303, 'mean_word_length': 4.767206477732794, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '436557', 'license': 'Apache-2.0', 'n_tokens_mistral': 958, 'n_tokens_neox': 947, 'n_words': 188, 'path': 'notebooks/05_Kubeflow_Pipeline/05_01_Pipeline_SDK_NodeSelectors.ipynb', 'repo_name': 'dalbhanj/eks-kubeflow-workshop', 'size': '2848'}