I had a successfully running pipeline but needed to update to create a datetime string to attach as a suffix to my model names, this value is important to reference the model later on in the pipeline.

I have this (truncated) code:

from kfp import compiler, dsl
from google_cloud_pipeline_components.v1.custom_job import CustomTrainingJobOp

@dsl.component(base_image="python:3.10")
def init() -> str:
    from datetime import datetime

    datetime_suffix = datetime.now().strftime("%Y%m%d-%H%M%S")
    return "datetime_suffix"


@dsl.pipeline(name="Training Pipeline", description="Training pipeline on Vertex AI")
def training_pipeline():
    init_op = init()
    datetime_suffix = init_op.output

    retrieval_job_name = "train-retrieval-model"
    train_retrieval_job = (
        CustomTrainingJobOp(
            project=PROJECT_ID,
            display_name=retrieval_job_name,
            location="us-west1",
            worker_pool_specs=[
                {
                    "machine_spec": {
                        "machine_type": "n1-standard-16",
                        "accelerator_type": "NVIDIA_TESLA_T4",  # See: https://cloud.google.com/vertex-ai/docs/training/configure-compute#gpu-compatibility-table
                        "accelerator_count": 1,
                    },
                    "replica_count": "1",
                    "container_spec": {
                        "image_uri": RETRIEVAL_TRAIN_IMAGE,
                        "env": [
                            {
                                "name": "ENTRYPOINT_SCRIPT",
                                "value": "trainer.retrieval.task",
                            },
                            {
                                "name": "ENVIRONMENT",
                                "value": "prod",
                            },
                            {
                                "name": "MODEL_SUFFIX",
                                "value": datetime_suffix,
                            },
                        ],
                    },
                }
            ],
        )
        .after(init_op)
        .set_display_name(retrieval_job_name)
    )

And I get the error:

Traceback (most recent call last):
  File "/Users/pthieu/python/commit/matchmaking/pipeline/main.py", line 93, in <module>
    @dsl.pipeline(name="Training Pipeline", description="Training pipeline on Vertex AI")
     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/dsl/pipeline_context.py", line 65, in pipeline
    return component_factory.create_graph_component_from_func(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/dsl/component_factory.py", line 673, in create_graph_component_from_func
    return graph_component.GraphComponent(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/dsl/graph_component.py", line 68, in __init__
    pipeline_spec, platform_spec = builder.create_pipeline_spec(
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 1919, in create_pipeline_spec
    build_spec_by_group(
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 1272, in build_spec_by_group
    subgroup_task_spec = build_task_spec_for_task(
                         ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 309, in build_task_spec_for_task
    to_protobuf_value(input_value))
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 78, in to_protobuf_value
    values=[to_protobuf_value(v) for v in value]))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 78, in <listcomp>
    values=[to_protobuf_value(v) for v in value]))
            ^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 74, in to_protobuf_value
    fields={k: to_protobuf_value(v) for k, v in value.items()}))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 74, in <dictcomp>
    fields={k: to_protobuf_value(v) for k, v in value.items()}))
               ^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 74, in to_protobuf_value
    fields={k: to_protobuf_value(v) for k, v in value.items()}))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 74, in <dictcomp>
    fields={k: to_protobuf_value(v) for k, v in value.items()}))
               ^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 78, in to_protobuf_value
    values=[to_protobuf_value(v) for v in value]))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 78, in <listcomp>
    values=[to_protobuf_value(v) for v in value]))
            ^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 74, in to_protobuf_value
    fields={k: to_protobuf_value(v) for k, v in value.items()}))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 74, in <dictcomp>
    fields={k: to_protobuf_value(v) for k, v in value.items()}))
               ^^^^^^^^^^^^^^^^^^^^
  File "/Users/pthieu/python/commit/matchmaking/pipeline/venv/lib/python3.11/site-packages/kfp/compiler/pipeline_spec_builder.py", line 80, in to_protobuf_value
    raise ValueError('Value must be one of the following types: '
ValueError: Value must be one of the following types: str, int, float, bool, dict, and list. Got: "{{channel:task=init;name=Output;type=String;}}" of type "<class 'kfp.dsl.pipeline_channel.PipelineParameterChannel'>".

Looks like CustomTrainingJobOp is not interpreting the returned value properly?

Here are my package versions:

google-cloud-pipeline-components==2.14.1
kfp==2.7.0
kfp-pipeline-spec==0.3.0
kfp-server-api==2.0.5

Note that I’m importing google-cloud-pipeline-components.v1 as I googled and people are saying there isn’t a proper equivalent in V2.