#define a function to add 2 numbers
def add(a: float, b: float) -> float:
  return a + b
  
  

import kfp.components as comp
add_op = comp.func_to_container_op(add) #  a factory function that you can use to create kfp.dsl.ContainerOp class instances for your pipeline
#function to write a float to a path
def write(path: str, x: float) -> str:
    num = str(x)
    f =open(path, "a")
    f.write(num)
    f.close
    return path

write("demofile2.txt",1) # testing our function


def read(path: str):
    f=open(path,"r")
    print(f.read())
    f.close


write_op = comp.func_to_container_op(write) # converting our function to a ContainerOp
read_op = comp.func_to_container_op(read) # converting our function to a ContainerOp

# new pipeline with volume being passed.
@dsl.pipeline(
    name='Volume Pipeline',
    description='simple pipeline to create a FRESH volume, add some numbers, and attach that volume on our different pods for their runs'
)

def volume_pipeline(
    a='1',
    b='2',
    c='3',
):

    vop = dsl.VolumeOp(
        name="volume_creation",
        resource_name="mypvc",
        size="5Gi",
        modes=dsl.VOLUME_MODE_RWM
    )
    add_task1 = add_op(a,3).add_pvolumes({"/mnt": vop.volume})
    add_task2 = add_op(add_task1.output,b)
    add_task3 = add_op(add_task2.output,c)  
    write_task=write_op("/mnt/output.txt",add_task3.output).add_pvolumes({"/mnt": vop.volume})
    read_task=read_op(write_task.output).add_pvolumes({"/mnt": vop.volume})

from typing import NamedTuple
import kfp


@kfp.components.create_component_from_func
def metadata_and_metrics() -> NamedTuple(
    "Outputs",
    [("mlpipeline_ui_metadata", "UI_metadata"), ("mlpipeline_metrics", "Metrics")],
):
    metadata = {
        "outputs": [
            {"storage": "inline", "source": "this should be bold", "type": "markdown"}
        ]
    }
    metrics = {
        "metrics": [
            {
                "name": "train-accuracy",
                "numberValue": 0.9,
            },
            {
                "name": "test-accuracy",
                "numberValue": 0.7,
            },
        ]
    }
    from collections import namedtuple
    import json

    return namedtuple("output", ["mlpipeline_ui_metadata", "mlpipeline_metrics"])(
        json.dumps(metadata), json.dumps(metrics)
    )


@kfp.dsl.pipeline()
def pipeline():
    metadata_and_metrics()

The Run output tab shows the visualizations for all pipeline steps in the selected run. To open the tab in the Kubeflow Pipelines UI:

{
  "name": "mlpipeline-ui-metadata",
  "optional": true,
  "path": "/tmp/outputs/mlpipeline_ui_metadata/data",
  "s3": {
    "accessKeySecret": {
      "key": "accesskey",
      "name": "mlpipeline-minio-artifact"
    },
    "bucket": "mlpipeline",
    "endpoint": "minio-service.kubeflow:9000",
    "insecure": true,
    "key": "artifacts/allocation-tx-5gpvh/allocation-tx-5gpvh-170870440/mlpipeline-ui-metadata.tgz",
    "secretKeySecret": {
      "key": "secretkey",
      "name": "mlpipeline-minio-artifact"
    }
  }
}

{
  "name": "mlpipeline-ui-metadata",
  "optional": true,
  "path": "/tmp/outputs/mlpipeline_ui_metadata/data",
  "s3": {
    "key": "artifacts/markdown-pipeline-ldvxr/2022/06/01/markdown-pipeline-ldvxr-4134324342/mlpipeline-ui-metadata.tgz"
  }
}