Florent VanDeMoortele
12/09/2025, 5:24 PMMarvin
12/09/2025, 5:24 PMMarvin
12/09/2025, 5:27 PMprefect.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`:
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:
prefect deploy
- Deploy a specific deployment by name:
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.Florent VanDeMoortele
12/09/2025, 5:31 PMMarvin
12/09/2025, 5:31 PMMarvin
12/09/2025, 5:33 PMdeployments:
- 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)
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:
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.Florent VanDeMoortele
12/09/2025, 5:35 PMMarvin
12/09/2025, 5:35 PMMarvin
12/09/2025, 5:42 PMpip 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:
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:
# 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:
prefect worker start --pool my-cloud-run-v2
Docs: Work pools • Serverless (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)
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)
prefect deploy # interactive wizard, or provide an entrypoint
This creates/updates prefect.yaml, e.g.:
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 Python • prefect.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:
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:
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",
)Marvin
12/09/2025, 5:42 PMFlorent VanDeMoortele
12/09/2025, 6:01 PMMarvin
12/09/2025, 6:01 PMMarvin
12/09/2025, 6:04 PMdeployments:
- 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
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 conceptFlorent VanDeMoortele
12/10/2025, 8:49 AMMarvin
12/10/2025, 8:49 AMMarvin
12/10/2025, 8:49 AMflow.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.Florent VanDeMoortele
12/10/2025, 8:52 AMMarvin
12/10/2025, 8:52 AMMarvin
12/10/2025, 8:55 AMprefect 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):
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:
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).