https://prefect.io logo
y

YSF

07/18/2023, 8:50 PM
Hi all, How would I go about running a pipline that has to run different but dependent steps in different images? So for example I might have Step 1(task or flow) run in image 1, and then if successful Step 2 run in image 2? Here's a yaml I have so far:
Copy code
pull:
- prefect.projects.steps.git_clone_project:
    repository: myscmprovider/myrepo.git
    branch: main
    access_token: '{{ prefect.blocks.secret.token }}'
- prefect.deployments.steps.set_working_directory:
    id: chdir
    directory: /opt/prefect/myrepo/k8-multi-container/

deployments:
  - name: deployment-1
    entrypoint: .\workflow\flow_demo.py:flow_1
    work_pool:
        name: aks-worker
        job_variables:
            image: myimage_1
            cpu_request: "4"
            cpu_limit: "6"
            memory_request: "4Gi"
            memory_limit: "6Gi"

  - name: deployment-2
    entrypoint: .\workflow\another_flow.py:flo_2
    work_pool:
        name: aks-worker
        job_variables:
            image: myimage_2
            cpu_request: "4"
            cpu_limit: "6"
            memory_request: "4Gi"
            memory_limit: "6Gi"
But I don't understand completely how these two deployments are linked? How does the dependency between flow 1 and flow 2 get established? They don't need to pass data, but flow 2 shouldn't run if flow 1 failed. And how would I kick it off all at once? Right now I do prefect deployments run to run things.
1
@Taylor Curran I think we spoke about this before, I did the PACC course which helped a lot. Can you help me understand where/how that dependency is established in this multi-container flow?
n

Nate

07/18/2023, 9:42 PM
hi @YSF - you wouldnt establish the dependency in the definition of the deployment, you would do that in the implementation of a flow that conditionally invokes those deployments with
run_deployment
for example
Copy code
from prefect import flow
from prefect.client.schemas.core import FlowRun
from prefect.deployments import run_deployment

@flow
def parent_flow():
   flow_run_1: FlowRun = run_deployment("flow-1/deployment-1")

   if flow_run_1.state.is_completed():
      flow_run_2: FlowRun = run_deployment("flow-2/deployment-2")
🙌 2
y

YSF

07/20/2023, 6:59 PM
Hey @Nate thank you! This makes sense, it threw me a no module found named prefect.client.schemas.core, and I went to the api reference and the page is just empty back until 2.10.8? Is this being deprecated? Or does it just live somewhere else now? https://docs.prefect.io/2.10.21/api-ref/prefect/client/schemas/
Actually, answered my own question, it lives under prefect.server now, so prefect.server.core.schemas
n

Nate

07/20/2023, 7:25 PM
whoops, i was just typing that in slack so there was no LSP to catch me I would use the client's schema
Copy code
from prefect.client.schemas.objects import FlowRun
good catch on the docs, will get that fixed
2 Views