[sdk] Cannot set dynamic node_selector or cpu/memory requests/limits

[TristanGreathouse]

Environment

• KFP version:
  Built from 1.8 branch the day before 1.8-rc2 was released.

• KFP SDK version:

kfp                        2.3.0                  
kfp-kubernetes             1.0.0                  
kfp-pipeline-spec          0.2.2                  
kfp-server-api             2.0.2

Steps to reproduce

I cannot define compute resources in a parent component and pass to a child component using the V2 SDK, although this was supported in V1 SDK versions. The problem can be reproduced by attempting to compile the following pipeline.

import os
import kfp
from kfp import dsl
from typing import NamedTuple
from kfp import kubernetes as kfp_k8s

@dsl.component
def define_resources() -> NamedTuple(
        'Outputs', 
        [
            ("nodetype", str),
            ("cpu_request", str),
            ("memory_request", str),
            ("env_var", str)
        ]
    ):
    import json
    from collections import namedtuple
    output_dict = {
        "nodetype": "xlarge",
        "cpu_request": "4",
        "memory_request": "16G"
    }
    resources = namedtuple("outputs", output_dict)
    return resources(**output_dict)


@dsl.component
def downstream() -> float:
    return 1.1


@dsl.pipeline
def compile_pipeline():
    define_resources_component = define_resources()
    downstream_component = downstream()
    kfp_k8s.add_node_selector(downstream_component, "type", define_resources_component.outputs["nodetype"])
    downstream_component.set_cpu_request(define_resources_component.outputs["cpu_request"])
    downstream_component.set_memory_request(define_resources_component.outputs["memory_request"])

if __name__ == "__main__":
    kfp.compiler.Compiler().compile(compile_pipeline, "test_pipeline.yaml")

The error for add_node_selector is:

TypeError: {{channel:task=define-resources;name=nodetype;type=String;}} has type PipelineParameterChannel, but expected one of: bytes, unicode

The error for set_{cpu/memory}_{request/limit} is:

TypeError: expected string or bytes-like object

Expected result

To be able to compile and run a pipeline using dynamic compute resources generated from an upstream parent component. This is essentially for pipelines to be reusable across different size datasets. For example, my pipelines may be used with a dataset that has 100 unique observations and requires a very small AWS EC2 instance, but it also may be used with tens of thousands of unique observations, and require a much larger AWS EC2 instance.

In our pipelines, the first component reads in an external config source and allocates compute resources to all downstream components based on this configuration. In V2, we'd have to hard-code values into pipelines which severely limits the scope and usability of the pipeline.

Materials and Reference

It seems this issue has examples that pass, but it was several years ago at this point.

Impacted by this bug? Give it a 👍.

[github-actions]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[github-actions]

This issue has been automatically closed because it has not had recent activity. Please comment "/reopen" to reopen it.