<@ULVA73B9P> How can I deploy with prefect.yaml th...
# ask-marvin
f
@Marvin How can I deploy with prefect.yaml the same flow with different parameters?
m
thought for 181.5 seconds
Yes — in Prefect 2.x you can define multiple deployments for the same flow in a single
prefect.yaml
by adding multiple entries under
deployments
that share the same
entrypoint
but have different
parameters
(and optionally different schedules, tags, work queues, etc.). Example `prefect.yaml`:
Copy code
name: my-project
# optional build/push/pull steps here

deployments:
  - name: daily-prod
    entrypoint: flows/etl.py:process_data
    parameters:
      environment: "prod"
      batch_size: 1000
      timeout: 3600
    tags: ["prod", "daily"]
    work_pool:
      name: my-work-pool
      work_queue_name: prod

  - name: hourly-dev
    entrypoint: flows/etl.py:process_data
    parameters:
      environment: "dev"
      batch_size: 100
      timeout: 600
    tags: ["dev", "hourly"]
    work_pool:
      name: my-work-pool
      work_queue_name: dev

  # optional: add schedules per deployment
  # - schedules:
  #     - cron: "0 9 * * *"
  #       timezone: "America/New_York"
  #       active: true
Deploy them: - Deploy everything in the file:
Copy code
prefect deploy
- Deploy a specific deployment by name:
Copy code
prefect deploy --name daily-prod
prefect deploy --name hourly-dev
Notes: - The
parameters
map must match your flow’s signature in
flows/etl.py:process_data
. - Ensure the
work_pool.name
exists (and any
work_queue_name
you reference). - You can keep things DRY with YAML anchors if you want to reuse shared config (e.g., the same
work_pool
config across deployments). If you share your flow signature and how you’re running on GCP (e.g., GKE via Kubernetes work pool, Cloud Run via Docker image, etc.), I can tailor the
work_pool
section to your setup.
f
@Marvin is it possible to deploy a flow with different schedule and different parameters for each schedule?
m
thought for 134.6 seconds
Short answer: In Prefect 2.x, parameters are set at the deployment level, not per schedule. You can attach multiple schedules to a single deployment, but they will all use the same parameters. If you need different parameters for different schedules, create separate deployments for each schedule + parameter set. Two ways to set this up in prefect.yaml: - Same parameters, multiple schedules (single deployment)
Copy code
deployments:
  - name: my-flow-all-schedules
    entrypoint: flows/etl.py:process_data
    parameters:
      mode: "standard"
      batch_size: 500
    schedules:
      - cron: "0 8 * * *"      # 8am
        timezone: "America/New_York"
        active: true
      - cron: "0 20 * * *"     # 8pm
        timezone: "America/New_York"
        active: true
    work_pool:
      name: my-work-pool
- Different parameters per schedule (multiple deployments of the same flow)
Copy code
deployments:
  - name: my-flow-morning
    entrypoint: flows/etl.py:process_data
    parameters:
      mode: "morning"
      batch_size: 1000
    schedules:
      - cron: "0 8 * * *"
        timezone: "America/New_York"
        active: true
    work_pool:
      name: my-work-pool

  - name: my-flow-evening
    entrypoint: flows/etl.py:process_data
    parameters:
      mode: "evening"
      batch_size: 200
    schedules:
      - cron: "0 20 * * *"
        timezone: "America/New_York"
        active: true
    work_pool:
      name: my-work-pool
Notes: - Each deployment must have a unique
name
. They can share the same
entrypoint
. - Make sure the
work_pool.name
exists in your Prefect Cloud workspace. - You can deploy both with:
Copy code
prefect deploy --name my-flow-morning --name my-flow-evening
If you’re open to upgrading: newer Prefect versions are adding support for per-schedule parameters within a single deployment. On 2.x today, separate deployments are the practical way to get different schedules with different parameters. If you share your GCP runtime (Kubernetes vs. Cloud Run, etc.), I can tailor the
work_pool
section for your setup.
f
@Marvin If I upgrade to prefect 3, how can I do that?
m
thought for 428.4 seconds
Great question — here’s a practical path to upgrade from Prefect 2 to Prefect 3 on Cloud (tailored for GCP). High‑level plan - Upgrade your runtime environments to Prefect 3, then migrate deployments to work pools/workers, update storage, and re‑deploy. - You can run Prefect 2 agents and Prefect 3 workers side‑by‑side during the transition. Step-by-step 1) Prep and package upgrades - Pause schedules for deployments you’ll migrate. - Upgrade in each execution environment (local, CI, Docker images, etc.):
Copy code
pip install -U prefect
pip install -U prefect-gcp  # if you use GCP blocks or Cloud Run/Vertex
prefect block register -m prefect_gcp
- Log back into Cloud if needed:
Copy code
prefect cloud login
prefect cloud workspace set --workspace "<account>/<workspace>"
2) Create GCP work pools (Prefect 3 replaces agents with workers/work pools) - Pick your target: - Hybrid (needs a worker process):
cloud-run
or
cloud-run-v2
- Serverless/push (no worker, Prefect provisions Cloud Run Jobs):
cloud-run:push
or
cloud-run-v2:push
- Create a work pool:
Copy code
# Cloud Run v2 hybrid pool (recommended if you prefer running workers)
prefect work-pool create my-cloud-run-v2 --type cloud-run-v2

# Or: Cloud Run v2 push pool (serverless)
prefect work-pool create my-cloud-run-v2-push --type cloud-run-v2:push
prefect work-pool provision-infrastructure my-cloud-run-v2-push
- If using a hybrid pool, start a worker where you want flow runs to execute:
Copy code
prefect worker start --pool my-cloud-run-v2
Docs: Work poolsServerless (push) pools 3) Recreate deployments (3.x) - Important removals: don’t use
Deployment.build_from_flow()
or
prefect deployment build
. In 3.x use
flow.deploy(...)
or
prefect deploy
. Option A: Deploy via Python (good for single flows)
Copy code
from prefect import flow

@flow(log_prints=True)
def my_flow(name: str = "world"):
    print(f"Hello {name}!")

if __name__ == "__main__":
    my_flow.deploy(
        name="my-flow-prod",
        work_pool_name="my-cloud-run-v2",   # your pool name
        image="<http://gcr.io/<project>/<image>:<tag>|gcr.io/<project>/<image>:<tag>>",  # if you use containers
    )
Option B: Deploy via prefect.yaml (good for teams/multiple flows)
Copy code
prefect deploy  # interactive wizard, or provide an entrypoint
This creates/updates
prefect.yaml
, e.g.:
Copy code
deployments:
  - name: my-flow-prod
    entrypoint: flows/my_flow.py:my_flow
    work_pool:
      name: my-cloud-run-v2
    pull:
      - prefect.deployments.steps.git_clone:
          repository: <https://github.com/<org>/<repo>.git>
          # For private repos, configure a credentials block/secret
Docs: Deploy via Pythonprefect.yaml 4) Update storage patterns (GitHub storage block was removed) - Replace 2.x GitHub storage block with git-based source or GCS. - Git-based example:
Copy code
from prefect import flow
from prefect.runner.storage import GitRepository
from prefect_github import GitHubCredentials

repo = GitRepository(
    url="<https://github.com/<org>/<repo>.git>",
    credentials=GitHubCredentials.load("my-github-creds")
)

flow.from_source(
    source=repo,
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="my-flow-prod",
    work_pool_name="my-cloud-run-v2",
)
- GCS example:
Copy code
from prefect import flow
from prefect_gcp.cloud_storage import GcsBucket
from prefect_gcp import GcpCredentials

gcs = GcsBucket(
    bucket="my-bucket",
    project="my-project",
    gcp_credentials=GcpCredentials.load("my-gcp-creds"),
)

flow.from_source(
    source=gcs,
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="my-flow-prod",
    work_pool_name="my-cloud-run-v2",
)
Docs: Store flow codeprefect-gcp integrationprefect-github integration 5) Roll out safely - Create new 3.x deployments alongside your 2.x ones. - Test manual runs. - Move schedules to the 3.x deployments; then stop the 2.x agent and remove the old deployments. 6) Code considerations in 3.x - Flow failure semantics: a flow only fails on an uncaught exception or if you return a failed state. If you relied on “any task failure fails the flow,” raise on failure explicitly. - Pydantic 2: if you use custom Pydantic models in parameters/blocks, scan for minor updates. Official guides - Upgrade to Prefect 3 - Agents to Workers - WorkersWork pools - Deploy via PythonDeploy via YAML - Store flow code - Serverless (push) pools - prefect-gcp integration A few quick questions so I can tailor exact steps/snippets: - Which GCP runtime are you using today: Cloud Run, GKE, Compute Engine, or local runners? - Where does your flow code live: GitHub, GCS, or baked into images? - Do you want fully serverless (Cloud Run push pool) or keep a worker process (hybrid pool)? - Any flows that rely on task failures automatically failing the flow?
f
@Marvin I don't want to know how to migrate to Prefect 3, but how to deploy a flow with different schedule and different parameters for each schedule in the prefect.yaml if I was in Prefect 3?
m
thought for 146.0 seconds
Absolutely — in Prefect 3 you can define multiple schedules for a single deployment and give each schedule its own parameters directly in prefect.yaml. Schedule-level parameters override the deployment’s default parameters. Example prefect.yaml
Copy code
deployments:
  - name: send-emails
    entrypoint: flows/send_email.py:send_email
    work_pool:
      name: my-pool

    # Deployment-level default parameters
    parameters:
      recipient: "<mailto:default@example.com|default@example.com>"
      message: "Hello"
      priority: "normal"

    # Multiple schedules with distinct parameters
    schedules:
      - cron: "0 8 * * *"
        slug: "morning-alert"
        timezone: "America/New_York"
        active: true
        parameters:
          recipient: "<mailto:morning-team@example.com|morning-team@example.com>"
          message: "Good morning!"
          priority: "high"

      - cron: "12 * * * *"
        slug: "hourly-check"
        timezone: "UTC"
        parameters:
          recipient: "<mailto:ops@example.com|ops@example.com>"
          # message/priority inherit from deployment defaults

      - interval: 86400   # seconds (24h)
        slug: "daily-summary"
        timezone: "America/Chicago"
        parameters:
          recipient: "<mailto:analytics@example.com|analytics@example.com>"
          message: "Daily report"
          priority: "low"
How parameters resolve - Start with deployment-level
parameters
- Merge schedule-level
parameters
for the specific schedule (these override defaults) - Manual runs use deployment-level parameters unless you provide overrides at run time Deploy it
Copy code
prefect deploy
# or, if using a custom file name:
prefect deploy --prefect-file prefect.yaml
Notes - Use
slug
and
timezone
per schedule for clarity and DST safety -
interval
is specified in seconds - Parameters must be JSON-serializable Docs - Create schedules - Deployment schema - Schedules concept
f
@Marvin are you sure this is not possible with Prefect 2?
m
thought for 28.6 seconds
Could you clarify what “this” refers to? Happy to double‑check, but the answer depends on the exact capability. If you’re asking about one of these common 3.x features: -
flow.from_source(...).deploy(...)
→ In Prefect 2.x, the equivalent is using deployments via the CLI or Python API: - CLI: build/apply a deployment (2.x only) - Python:
Deployment.build_from_flow(...)
(2.x only) -
prefect deploy
CLI (3.x) → In 2.x you use
prefect deployment build
+
prefect deployment apply
- GitHub “storage” via
.from_source()
(3.x) → In 2.x you’d use storage blocks (e.g., GitHub/GCS) on your deployment Since you’re on Prefect 2 Cloud and GCP, I can outline the exact 2.x steps once I know: - What Prefect 2 version you’re on (e.g., 2.10+ uses work pools/workers; older versions use agents/queues) - The specific feature you want (e.g., deploying from a Git repo, running on GKE/Cloud Run, passing secrets, etc.) - Any code or command you tried and the outcome Share those details and I’ll translate it to the correct Prefect 2 approach.
f
@Marvin This refers to have multiples schedules for a deployment in the prefect.yaml file and have differents parameters for each schedule
m
thought for 218.2 seconds
Short answer: In Prefect 2.x, you can attach multiple schedules to a single deployment, but you cannot give each schedule its own parameters. All schedules on a deployment use the deployment’s parameters. What you can do in 2.x: - Multiple schedules on one deployment (same params for all schedules) - In UI: add multiple schedules on the deployment page - In CLI:
prefect deployment schedule create ...
- In YAML: depending on your 2.x version,
schedules:
may be supported for worker-based deployments; if not, add them via UI/CLI - Different parameters per schedule → create separate deployments (one per schedule), each with its own
parameters
and
schedule
Examples - One deployment, multiple schedules (same parameters):
Copy code
deployments:
- name: my-deployment
  entrypoint: flows/my_flow.py:my_flow
  parameters:
    mode: "default"
  schedules:
    - cron: "0 8 * * *"
      timezone: "UTC"
      active: true
    - cron: "0 18 * * *"
      timezone: "UTC"
      active: true
- Multiple deployments to vary parameters per schedule:
Copy code
deployments:
- name: my-deployment-morning
  entrypoint: flows/my_flow.py:my_flow
  parameters:
    mode: "morning"
  schedule:
    cron: "0 8 * * *"
    timezone: "UTC"

- name: my-deployment-evening
  entrypoint: flows/my_flow.py:my_flow
  parameters:
    mode: "evening"
  schedule:
    cron: "0 18 * * *"
    timezone: "UTC"
Docs - Schedules in Prefect 2: Schedules (Prefect 2) - Deployments and prefect.yaml: Deployments (Prefect 2) If you share your current
prefect.yaml
and the schedules/params you want, I can translate it into the recommended 2.x setup (either multiple schedules on one deployment or multiple deployments to vary parameters).