gigl.orchestration.kubeflow.kfp_orchestrator#

Attributes#

`DEFAULT_KFP_COMPILED_PIPELINE_DEST_PATH`
`DEFAULT_PIPELINE_VERSION_NAME`
`DEFAULT_START_AT_COMPONENT`
`logger`

Classes#

KfpOrchestrator

Orchestration of Kubeflow Pipelines for GiGL.

Module Contents#

class gigl.orchestration.kubeflow.kfp_orchestrator.KfpOrchestrator[source]#

Orchestration of Kubeflow Pipelines for GiGL. .. method:: compile

Compiles the Kubeflow pipeline.

run()[source]#

Runs the Kubeflow pipeline.

Parameters:

applied_task_identifier (gigl.src.common.types.AppliedTaskIdentifier)
task_config_uri (gigl.common.Uri)
resource_config_uri (gigl.common.Uri)
start_at (str)
stop_after (Optional[str])
compiled_pipeline_path (gigl.common.Uri)

Return type:

google.cloud.aiplatform.PipelineJob

upload()#: Uploads the pipeline to KFP.

wait_for_completion()[source]#

Waits for the pipeline run to complete.

Parameters:: run (Union[google.cloud.aiplatform.PipelineJob, str])

classmethod compile(cuda_container_image, cpu_container_image, dataflow_container_image, dst_compiled_pipeline_path=DEFAULT_KFP_COMPILED_PIPELINE_DEST_PATH, additional_job_args=None, tag=None)[source]#

Compiles the GiGL Kubeflow pipeline.

Parameters:

cuda_container_image (str) – Container image for CUDA (see: containers/Dockerfile.cuda).
cpu_container_image (str) – Container image for CPU.
dataflow_container_image (str) – Container image for Dataflow.
dst_compiled_pipeline_path (Uri) – Destination path for the compiled pipeline YAML file. Defaults to
additional_job_args (Optional[dict[gigl.src.common.constants.components.GiGLComponents, dict[str, str]]])
tag (Optional[str])

Return type:

gigl.common.Uri

:param DEFAULT_KFP_COMPILED_PIPELINE_DEST_PATH.: :param additional_job_args: Additional arguments to be passed into components, organized by component. :type additional_job_args: Optional[dict[GiGLComponents, dict[str, str]]] :param tag: Optional tag to include in the pipeline description. :type tag: Optional[str]

Returns:

The URI of the compiled pipeline.

Return type:

Uri

Parameters:

cuda_container_image (str)
cpu_container_image (str)
dataflow_container_image (str)
dst_compiled_pipeline_path (gigl.common.Uri)
additional_job_args (Optional[dict[gigl.src.common.constants.components.GiGLComponents, dict[str, str]]])
tag (Optional[str])

run(applied_task_identifier, task_config_uri, resource_config_uri, start_at=DEFAULT_START_AT_COMPONENT, stop_after=None, compiled_pipeline_path=DEFAULT_KFP_COMPILED_PIPELINE_DEST_PATH)[source]#

Runs the GiGL Kubeflow pipeline.

Parameters:

applied_task_identifier (AppliedTaskIdentifier) – Identifier for the task.
task_config_uri (Uri) – URI of the task configuration file.
resource_config_uri (Uri) – URI of the resource configuration file.
start_at (str) – Component to start the pipeline at. Defaults to ‘config_populator’.
stop_after (Optional[str]) – Component to stop the pipeline after. Defaults to None i.e. run entire pipeline.
compiled_pipeline_path (Uri) – Path to the compiled pipeline YAML file.

Returns:

The created pipeline job.

Return type:

aiplatform.PipelineJob

wait_for_completion(run)[source]#

Waits for the completion of a pipeline run.

Parameters:: run (Union[aiplatform.PipelineJob, str]) – The pipeline job or its resource name.
Returns:: None

gigl.orchestration.kubeflow.kfp_orchestrator.DEFAULT_KFP_COMPILED_PIPELINE_DEST_PATH[source]#

gigl.orchestration.kubeflow.kfp_orchestrator.DEFAULT_PIPELINE_VERSION_NAME = 'gigl-pipeline-version-at-Uninferable'[source]#

gigl.orchestration.kubeflow.kfp_orchestrator.DEFAULT_START_AT_COMPONENT = 'config_populator'[source]#

gigl.orchestration.kubeflow.kfp_orchestrator.logger[source]#