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]])
fail_fast (bool)

Return type:

None

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, exit_task=ExitTask.EMAIL)[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])
exit_task (gigl.orchestration.kubeflow.kfp_pipeline.ExitTask)

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] :param exit_task: What the pipeline runs when it exits. See ExitTask. :type exit_task: ExitTask

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])
exit_task (gigl.orchestration.kubeflow.kfp_pipeline.ExitTask)

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, fail_fast=False)[source]#

Waits for the completion of one or more pipeline runs.

Parameters:

run (Union[aiplatform.PipelineJob, str]) – The pipeline job or its resource name.
fail_fast (bool) – If True, the first failed run cancels every run that is still waiting, then this waits for Vertex AI to report those runs as cancelled, which takes a few minutes. A run whose status request failed also gets a cancel, because it can still be running. A cancelled run compiled with ExitTask.CLEANUP, as the e2e tests are, cancels its own Vertex AI, Dataflow, and Dataproc child jobs in its exit task; other runs leave their child jobs running. If False, every run is awaited. Either way the error names each failed run.

Raises:

RuntimeError – If any run failed, or a cancel could not be requested. The message names every failed run and every failed cancel with its error. A run is excused only if this call cancelled it and it then reached CANCELLED.

Return type:

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