Kubeflow Pipeline DSL helper

kfx.dsl is a module with helper classes to config some of the more common settings for the kfp.dsl.ContainerOp objects in the pipeline.

kfx.dsl._transformers.ContainerOpTransform

Helper class to manipulate some common internal properties of ContainerOp.

Please refer to documentation for full set of transforms available.

::

import kfp.components
import kfp.dsl
import kfx.dsl

transforms = (
    kfx.dsl.ContainerOpTransform()
    .set_resources(cpu="500m", memory=("1G", "4G"))
    .set_image_pull_policy("Always")
    .set_env_vars({"ENV": "production"})
    .set_env_var_from_secret("AWS_ACCESS_KEY", secret_name="aws", secret_key="access_key")
    .set_annotations({"iam.amazonaws.com/role": "some-arn"})
)


@kfp.dsl.components.func_to_container_op
def echo(text: str) -> str:
    print(text)
    return text


@kfp.dsl.pipeline(name="demo")
def pipeline(text: str):
    op1 = echo(text)
    op2 = echo("%s-%s" % text)

    # u can apply the transform on op1 only
    # op1.apply(transforms)

    # or apply on all ops in the pipeline
    kfp.dsl.get_pipeline_conf().add_op_transformer(transforms)

Methods

__init__(self, transforms=None) special

Creates a new instance of ContainerOpTransform object.

Parameters:

Name	Type	Description	Default
transforms	List[Callable[[kfp.dsl._container_op.ContainerOp], kfp.dsl._container_op.ContainerOp]]	Optional list of custom transform functions. Defaults to None.	None

add_env_var(self, name, value)

Update the transform function to set the provided env var to the ContainerOp.

Parameters:

Name	Type	Description	Default
name	str	name of the env var.	required
value	str	value of the env var.	required

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

add_env_var_from_configmap(self, configmap_name)

Update the transform function to set env vars from a configmap.

Parameters:

Name	Type	Description	Default
configmap_name	str	name of the configmap.	required

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

add_env_var_from_secret(self, name, secret_name, secret_key)

Update the transform function to set the provided env var from a k8s secret.

Parameters:

Name	Type	Description	Default
name	str	name of the env var.	required
secret_name	str	name of the k8s secret.	required
secret_key	str	key to retrieve from the k8s secret.	required

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

add_env_vars(self, env_vars)

Update the transform function to set the provided env vars to the ContainerOp.

Parameters:

Name	Type	Description	Default
env_vars	Dict[str, str]	dict of env vars keys and values.	required

set_annotations(self, annotations)

Update the transform function to set the provided annotations to the ContainerOp.

Parameters:

Name	Type	Description	Default
annotations	Dict[str, str]	dict of annotation keys and values.	required

set_cpu_resources(self, request, limit=None)

Update the transform function to set the cpu resources for the main container.

Parameters:

Name	Type	Description	Default
request	Union[int, str]	How much cpu to request.	required
limit	Union[int, str]	Max cpu load before throttling. Defaults to None.	None

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

set_gpu_limit(self, value, vendor='nvidia')

Update the transform function to set the cpu limit for the main container.

Parameters:

Name	Type	Description	Default
value	Union[int, str]	GPU limit for the main container.	required
vendor	str	Either "nvidia" or "amd". Defaults to "nvidia".	'nvidia'

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

set_image_pull_policy(self, policy)

Update the transform function to set the image pull policy for the main container.

Parameters:

Name	Type	Description	Default
policy	str	One of "Always", "Never", "IfNotPresent".	required

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

set_labels(self, labels)

Update the transform function to set the provided labels to the ContainerOp.

Parameters:

Name	Type	Description	Default
labels	Dict[str, str]	dict of labels keys and values.	required

set_memory_resources(self, request, limit=None)

Update the transform function to set the memory resources for the main container.

Parameters:

Name	Type	Description	Default
request	Union[int, str]	How much memory to request.	required
limit	Union[int, str]	Max memory before killing the pod. Defaults to None.	None

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

set_resources(self, cpu=None, memory=None)

Update the transform function to set the cpu and memory resources for the main container.

If a int or str is provided, the resource request will equal the limit (i.e. QoS is Guaranteed). Otherwise, a tuple should be provided specifying the request and the limit.

Parameters:

Name	Type	Description	Default
cpu	Union[int, str, Tuple[Union[int, str], Union[int, str]]]	A str or tuple representing the cpu request and limit.	None
memory	Union[str, Tuple[str, str]]	A str or tuple representing the memory request and limit.	None

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

set_sidecar_image_pull_policy(self, policy, sidecar_name='*')

Update the transform function to set the image pull policy for the sidecars.

Parameters:

Name	Type	Description	Default
policy	str	One of "Always", "Never", "IfNotPresent".	required
sidecar_name	str	Glob pattern for sidecar name. Defaults to "*".	'*'

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

set_sidecar_resources(self, cpu=None, memory=None, sidecar_name='*')

Update the transform function to set the cpu and memory resources for the sidecars.

If a int or str is provided, the resource request will equal the limit (i.e. QoS is Guaranteed). Otherwise, a tuple should be provided specifying the request and the limit.

Parameters:

Name	Type	Description	Default
cpu	Union[int, str, Tuple[Union[int, str], Union[int, str]]]	A str or tuple representing the cpu request and limit.	None
memory	Union[str, Tuple[str, str]]	A str or tuple representing the memory request and limit.	None
sidecar_name	str	Glob pattern matching the sidecar name. Defaults to "*".	'*'

Returns:

Type	Description
ContainerOpTransform	ContainerOpTransform: updated ContainerOpTransform object.

---

Last update: June 9, 2021