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)

  • labels (Optional[dict[str, str]])

  • notification_emails (Optional[list[str]])

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[str, google.cloud.aiplatform.PipelineJob, list[str], list[google.cloud.aiplatform.PipelineJob]])

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:
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, labels=None, notification_emails=None)[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.

  • labels (Optional[dict[str, str]]) – Labels to associate with the run.

  • notification_emails (Optional[list[str]]) – Emails to send notification to. See https://cloud.google.com/vertex-ai/docs/pipelines/email-notifications for more details.

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]#