enml.models import PipelineDeploymentResponseModelfrom zenml.orchestrators import ContainerizedOrchestrator

from zenml.stack import Stack

class MyOrchestrator(ContainerizedOrchestrator):

def get_orchestrator_run_id(self) -> str:

# Return an ID that is different each time a pipeline is run, but the

# same for all steps being executed as part of the same pipeline run.

# If you're using some external orchestration tool like Kubeflow, you

# can usually use the run ID of that tool here.

...

def prepare_or_run_pipeline(

self,

deployment: "PipelineDeploymentResponseModel",

stack: "Stack",

environment: Dict[str, str],

) -> None:

# If your orchestrator supports scheduling, you should handle the schedule

# configured by the user. Otherwise you might raise an exception or log a warning

# that the orchestrator doesn't support scheduling

if deployment.schedule:

...

for step_name, step in deployment.step_configurations.items():

image = self.get_image(deployment=deployment, step_name=step_name)

command = StepEntrypointConfiguration.get_entrypoint_command()

arguments = StepEntrypointConfiguration.get_entrypoint_arguments(

step_name=step_name, deployment_id=deployment.id

# Your orchestration tool should run this command and arguments

# in the Docker image fetched above. Additionally, the container which

# is running the command must contain the environment variables specified

# in the `environment` dictionary.

# If your orchestrator supports parallel execution of steps, make sure

# each step only runs after all its upstream steps finished

upstream_steps = step.spec.upstream_steps

# You can get the settings your orchestrator like so.

# The settings are the "dynamic" part of your orchestrators config,

# optionally defined when you register your orchestrator but can be

# overridden at runtime.

# In contrast, the "static" part of your orchestrators config is

# always defined when you register the orchestrator and can be

# accessed via `self.config`.

step_settings = cast(