https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> How should I structure multiple Prefe...
# ask-marvin
w
@Marvin How should I structure multiple Prefect deployments that belong to the same project? Is it better to run them from one unified container environment, or isolate each deployment in its own container for better reliability and scalability?
m
✅ thought for 284.0 seconds
n
broadly speaking i'd suggest • 1 python package (e.g. https://github.com/zzstoatzz/prefect-pack) ◦ common utils etc • extras (optional-dependencies) or groups for different "classes" of python deps your flows need ◦ build docker images for each set of deps your flows need
m
Short answer: both patterns work in Prefect 3.x. Use one shared image/runtime when your flows share dependencies and SLAs; move to per-deployment images when you need isolation (conflicting deps), independent release cadence, or tighter security/resource guarantees. A practical rule of thumb - Start unified: one image, one work pool; use work queues and job variables to differentiate priority, resources, and config. - Split when needed: give each deployment its own image (and optionally its own pool) if you hit dependency conflicts, security boundaries, or noisy-neighbor issues. Pros and cons - One shared image/runtime - Pros: simplest CI/CD, fastest builds, fewer artifacts to manage - Cons: blast radius on any dependency change, coupled release cadence, less isolation of secrets/resources - Per-deployment images - Pros: dependency isolation, least-privilege security, independent versioning, easier targeted scaling - Cons: more images and CI jobs to maintain, longer/parallelized builds How to implement in Prefect 3.x - Shared image across many deployments - Create a single work pool (Docker/Kubernetes/etc.) and set a base job template/image. - Put deployments in different queues for priority and concurrency. - Use 
job_variables
per deployment for env/config/resources; leave image at the pool default. - Verified CLI
Copy code
prefect work-pool create my-docker-pool --type docker
    prefect work-queue create my-docker-pool/critical --priority 1 --limit 2
    prefect work-queue create my-docker-pool/batch --priority 10 --limit 10
- Example
Copy code
from prefect import flow

    @flow
    def etl(): ...

    # Shared image via pool; per-deployment config via job_variables and queues
    etl.deploy(
        name="etl-critical",
        work_pool_name="my-docker-pool",
        work_queue_name="critical",
        tags=["etl","prod"],
        concurrency_limit=2,
        job_variables={
            "env": {"PROFILE": "prod"},
            "limits": {"cpu": "1500m", "memory": "1Gi"},
        },
    )

    etl.deploy(
        name="etl-batch",
        work_pool_name="my-docker-pool",
        work_queue_name="batch",
        tags=["etl","batch"],
        job_variables={
            "env": {"PROFILE": "batch"},
            "limits": {"cpu": "500m", "memory": "512Mi"},
        },
    )
- Per-deployment isolated images - Keep a common base Dockerfile if helpful, but each deployment sets its own image. - You can stay on the same work pool; just override the image per deployment. Create separate pools only if the infrastructure differs (e.g., GPU nodes vs CPU). - Example
Copy code
from prefect import flow
    from prefect.infrastructure.docker import DockerImage

    @flow
    def train_model(): ...

    @flow
    def report_metrics(): ...

    train_model.deploy(
        name="train-model",
        work_pool_name="ml-pool",
        image="registry/my-ml-image:cuda-12.1",
        tags=["ml","gpu"],
        job_variables={
            "limits": {"cpu": "2000m", "memory": "8Gi"},
            "env": {"WANDB_MODE": "online"},
        },
    )

    report_metrics.deploy(
        name="report-metrics",
        work_pool_name="ml-pool",
        image=DockerImage(dockerfile="Dockerfile.metrics", buildargs={"PYPI_MIRROR":"https://..."}),
        tags=["reporting"],
        job_variables={"limits": {"cpu": "500m", "memory": "512Mi"}},
    )
Other knobs that help regardless of approach - Concurrency limits: set per deployment (
concurrency_limit
) and/or via queue limits to avoid overwhelming shared systems. - Schedules and tags: organize by environment, domain, cost center, etc. - From-source deployments: 
flow.from_source(...).deploy(...)
to pull code at runtime; handy when you want fewer images but still isolate runtime via job variables. - CI/CD: matrix builds for per-deployment images; single build for shared-image model.
- Security: keep credentials out of images; inject via Prefect blocks or 
job_variables.env
at run time. 3.x specifics to avoid surprises - Use 
flow.deploy(...)
or 
prefect deploy
in 3.x; 
Deployment.build_from_flow()
and 
prefect deployment build
were removed. - Agents are replaced by workers; use work pools/queues for routing and scaling. If you share a bit about: - How many deployments and how different their dependencies are - Your runtime (Docker vs. Kubernetes) and scaling needs - Any security/isolation requirements - Your current CI/CD setup …I can recommend a concrete layout (pools/queues/images) and give you a starter 
prefect.yaml
and Dockerfile(s) tailored to your project.
w
@Marvin What is the best way to associate and later look up Prefect flow runs using a business-level task_id — using tags, parameters, or some other mechanism?
m
✅ thought for 409.0 seconds
Great question. In Prefect 3.x, the most reliable and queryable way to associate a business-level ID (e.g., task_id) with flow runs is to use a combination of: - Flow run name for a primary identifier - Tags for additional, multi-valued classifications - Idempotency key when you launch runs programmatically (gives you both de-duplication and direct lookup) Avoid relying on parameters for lookup; they are not indexed for filtering. Recommended pattern - Put the business ID in the flow run name (easy to see, easy to search) - Add a normalized tag like task_id:<value> (good for filtering and analytics) - If you create runs via API or run_deployment, set idempotency_key to the business ID (ensures de-duplication and exact lookup) Examples 1) Set run name and tags from parameters in the flow
Copy code
from prefect import flow, tags
from prefect.artifacts import create_markdown_artifact

@flow(flow_run_name="task-{task_id}")
def process_task(task_id: str, customer_id: str, region: str):
    # Tags can carry multiple identifiers/classifications
    with tags(f"task_id:{task_id}", f"customer:{customer_id}", f"region:{region}"):
        # Optional: write an artifact for rich audit details
        create_markdown_artifact(
            key="business-summary",
            markdown=f"# Task {task_id}\nCustomer: {customer_id}\nRegion: {region}"
        )
        ...
2) Launch a run with name, tags, and idempotency key (API or a parent flow)
Copy code
from prefect.deployments import run_deployment

# This creates the run with the desired name/tags and enables deduplication
run = run_deployment(
    name="my-project/process-task",
    parameters={"task_id": "T-123", "customer_id": "C-456", "region": "us-east"},
    flow_run_name="task-T-123",
    tags=["task_id:T-123", "customer:C-456", "region:us-east"],
    idempotency_key="T-123",  # same incoming task_id if it must be unique
)
3) Query runs by name, tags, or idempotency key
Copy code
import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter, FlowRunFilterName, FlowRunFilterTags, FlowRunFilterIdempotencyKey
)

async def find_runs():
    async with get_client() as client:
        # By exact name
        by_name = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(name=FlowRunFilterName(any_=["task-T-123"]))
        )

        # By partial name
        like_name = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(name=FlowRunFilterName(like_="task-T-"))
        )

        # By tag
        by_tag = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(tags=FlowRunFilterTags(any_=["task_id:T-123"]))
        )

        # By idempotency key (best for exact “business ID → run” lookup)
        by_idem = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                idempotency_key=FlowRunFilterIdempotencyKey(any_=["T-123"])
            )
        )

    return by_name, like_name, by_tag, by_idem

asyncio.run(find_runs())
Notes and gotchas - Parameters: Great for internal logic, but not filterable. Don’t rely on them for lookup. - Run names: Highly visible and searchable. Use for your primary business ID. - Tags: Perfect for multiple attributes. If you update tags later via the client, be sure to read and merge existing tags first—updates replace the tag list. - Idempotency key: If you programmatically create runs (API/parent flow), setting this to your task_id gives you de-duplication and exact lookup via FlowRunFilterIdempotencyKey. If your business ID can repeat, don’t use idempotency_key. - Privacy: Tags and names are visible in the UI; avoid putting sensitive data directly. Consider hashing if needed. If you share how your runs are triggered (scheduled vs. API/parent) and whether the business task_id is unique, I can tailor this to your setup (e.g., naming conventions, tag schemas, and the exact client filters you’ll use).
w
@Marvin and what about using tags for flow? How can I provide tag from main flow run to child subflow?
m
✅ thought for 63.9 seconds
Yes—use tags for flows. In Prefect 3.x, tags applied via the tags context manager on a parent flow are inherited by: - all tasks created inside that context - any subflows you call inside that context So the simplest way to pass a tag from a main flow to a child subflow is to wrap the subflow call in a tags(...) block. Examples 1) Subflow inherits parent tags automatically
Copy code
from prefect import flow, tags

@flow
def child(task_id: str):
    ...

@flow
def parent(task_id: str):
    # Apply tags once; both tasks and subflows inside get these tags
    with tags(f"task_id:{task_id}", "region:us-east"):
        child(task_id)   # subflow run will include these tags
2) Add child-specific tags on top of the inherited ones
Copy code
from prefect import flow, tags

@flow
def child(task_id: str):
    with tags("child-phase:transform"):
        ...
3) If you start a “child” run via run_deployment (not a subflow), pass tags explicitly
Copy code
from prefect.deployments import run_deployment
from prefect import flow

@flow
def parent(task_id: str):
    run_deployment(
        name="my-project/child",
        parameters={"task_id": task_id},
        tags=[f"task_id:{task_id}", "region:us-east"]  # must pass explicitly
    )
Note: runs started with run_deployment do not automatically inherit the parent’s context tags—you need to pass them. 4) Modifying tags after a run starts (if needed) If you must add tags after creation, read and merge; updates replace the list.
Copy code
from prefect import flow, get_client
from prefect.context import get_run_context

@flow
async def child():
    rc = get_run_context()
    async with get_client() as client:
        existing = rc.flow_run.tags or []
        new_tags = list(set(existing + ["extra:tag"]))
        await client.update_flow_run(flow_run_id=rc.flow_run.id, tags=new_tags)
Best practices - Use tags context in the parent around the subflow call for automatic inheritance. - Standardize tag keys (e.g., task_id:<id>, region:<region>) for easy filtering. - For runs triggered via run_deployment, pass tags explicitly. - Avoid putting sensitive data in tag values (visible in UI).
w
@Marvin and how can I provide business parameter "task_id" from parent flow-run to child flow run? context, def parameter? What else?
m
✅ thought for 57.0 seconds
Great question. In Prefect 3.x there’s no automatic parameter inheritance between flows, so you should explicitly pass your business ID (“task_id”) to the child. Here are the reliable options, in order of preference: 1) Native subflow call (best) - Treat the child like a normal function and pass the parameter.
Copy code
from prefect import flow

@flow
def child(task_id: str):
    ...

@flow
def parent(task_id: str):
    child(task_id=task_id)  # explicit propagation
2) Child triggered via run_deployment (best when using deployments) - Pass parameters explicitly when you create the child run. By default, run_deployment links the child as a subflow (as_subflow=True), but parameters do not auto-propagate—you must pass them.
Copy code
from prefect.deployments import run_deployment
from prefect import flow

@flow
def parent(task_id: str):
    run_deployment(
        name="project/child",
        parameters={"task_id": task_id},
        # optional extras:
        flow_run_name=f"task-{task_id}",
        tags=[f"task_id:{task_id}"],
    )
3) Defensive fallback inside the child (inherit from parent if not provided) - If you want the child to “pick up” the parent’s task_id when the caller forgets to pass it, read the parent run’s parameters via runtime context. This works for both native subflows and runs created via run_deployment (as_subflow=True), because the child has parent_flow_run_id set.
Copy code
from typing import Optional
from prefect import flow, get_client
from prefect.runtime import flow_run as rt_flow_run

@flow
async def child(task_id: Optional[str] = None):
    if not task_id:
        parent_id = rt_flow_run.parent_flow_run_id
        if parent_id:
            async with get_client() as client:
                parent = await client.read_flow_run(parent_id)
                task_id = parent.parameters.get("task_id")
    # use task_id from here on
4) Tags as a last-resort transport - If the parent already adds a tag like task_id:T-123, the child (as a subflow) will inherit tags. You could parse the ID from tags in the child—but this is brittle. Prefer parameters for data, tags for classification.
Copy code
from prefect import flow
from prefect.runtime import flow_run as rt_flow_run

@flow
def child():
    task_id = None
    for t in rt_flow_run.tags or []:
        if t.startswith("task_id:"):
            task_id = t.split(":", 1)[1]
            break
What not to use for per-run IDs - Variables/Blocks: good for global/shared config, not for run-scoped identifiers like task_id. Summary - Native subflow: pass task_id as an argument. - run_deployment: pass task_id via parameters in the API call. - Optional fallback: child reads parent’s parameters via prefect.runtime.flow_run.parent_flow_run_id + client. - Avoid relying on tags for passing data; use them as a complement for search and filtering.
n
there are million ways you can do global state, context vars, redis, Prefect Variables. but passing parameters i often find simplest/best
w
@Marvin What is the best strategy for logging in Prefect 3.x? Should I call 
get_run_logger()
in every flow and task, or is there a more efficient or recommended way to structure logging?
m
✅ thought for 70.0 seconds
Short answer: use get_run_logger inside each flow/task, but do it once per function and reuse the local logger. Let Prefect capture any standard logging from libraries automatically. Don’t try to create/run-scope loggers at module import time or pass logger objects between tasks. Recommended patterns - In flows and tasks: call once and reuse
Copy code
from prefect import flow, task, get_run_logger

  @task
  def compute(x: int) -> int:
      logger = get_run_logger()
      <http://logger.info|logger.info>("Computing x=%s", x)
      return x * 2

  @flow
  def my_flow():
      logger = get_run_logger()
      <http://logger.info|logger.info>("Starting flow")
      y = compute(21)
      <http://logger.info|logger.info>("Result=%s", y)
- Let third‑party logs be auto-captured - Prefect’s logging config attaches handlers so standard Python logging (e.g., logging.getLogger(name).info(...)) emitted during a flow/task run is captured and sent to Prefect automatically. - Use get_run_logger when you want Prefect’s run context in the message (flow/task run IDs, nice formatting in UI). - Don’t use module-level get_run_logger - Avoid 
logger = get_run_logger()
at import time; there’s no run context then. Always call it inside the flow/task body. - Don’t pass loggers across task boundaries - Tasks may run in separate processes/containers. Just call get_run_logger in each task; it’s lightweight. - If you trigger child runs or subflows - Each subflow/task gets its own run-scoped logger. Call get_run_logger in that subflow/task as needed. Nice-to-haves - Redirect prints to logs if you have legacy code:
Copy code
from prefect.logging import patch_print

  @flow
  def my_flow():
      patch_print()  # sends print() to the run logger
      print("This will appear as a Prefect log")
- Add business context without polluting tags:
Copy code
logger = get_run_logger()
  <http://logger.info|logger.info>("Processing", extra={"task_id": task_id})
Note: extra fields are great for structured sinks; the UI primarily displays the message text. - Tune globally via Prefect settings/profile - Configure log level/format/handlers via your Prefect profile or settings file. Keep code simple and let configuration control verbosity. Rule of thumb - Call get_run_logger at the top of each flow/task and reuse it. - Use standard logging in libraries/utilities; Prefect will capture it during runs. - Avoid global run loggers and passing logger objects around.
w
@Marvin Could you suggest the best way to create a task in Prefect 3.0 so that a failure of this task (for example, creating an artifact when an external storage is unavailable) does not cause the entire flow to fail, and the flow can continue executing other tasks?
m
✅ thought for 346.6 seconds
Great question. In Prefect 3.x there are a few clean ways to make a “best-effort” task (like creating an artifact) that won’t take down your flow if it fails. Key behavior to know: - A task failure does not automatically fail the flow in Prefect 3.x. However, downstream tasks that depend on a failed task will not start unless you explicitly allow it. - Use 
allow_failure(...)
to let downstream tasks run even when an upstream failed. Recommended patterns 1) Keep failures visible, but don’t block downstream: use allow_failure - Let your artifact task raise if it can’t create the artifact (so you can see a Failed task in the UI). - For any downstream that shouldn’t be blocked by the artifact, wrap the upstream future/state with 
allow_failure(...)
. The downstream will receive either the successful result or the Exception.
Copy code
from prefect import flow, task, allow_failure
from prefect.artifacts import create_markdown_artifact
from prefect import get_run_logger

@task
def publish_artifact(text: str) -> str:
    # Will raise if external storage is unavailable
    return create_markdown_artifact(markdown=text, key="daily-report")

@task
def continue_work(artifact_outcome):
    logger = get_run_logger()
    if isinstance(artifact_outcome, Exception):
        logger.warning(f"Artifact creation failed (non-blocking): {artifact_outcome}")
        # proceed without the artifact
        return "continued without artifact"
    else:
        <http://logger.info|logger.info>(f"Artifact created: {artifact_outcome}")
        return "continued with artifact"

@flow
def main():
    art = publish_artifact.submit("# Hello from Prefect")
    # This downstream will run even if the artifact task failed
    continue_work.submit(allow_failure(art))

    # Other independent tasks will also continue running as usual
Notes: - 
allow_failure(future)
passes the Exception object to the downstream if the upstream failed; otherwise it passes the successful result. - This keeps the artifact task marked as Failed if it truly failed, but your flow and other tasks keep going. 2) Inspect a task’s State explicitly: return_state=True - If you want to branch on the actual state (and log messages, etc.):
Copy code
from prefect import flow, task, allow_failure, get_run_logger
from prefect.artifacts import create_markdown_artifact

@task
def publish_artifact(text: str) -> str:
    return create_markdown_artifact(markdown=text, key="daily-report")

@task
def handle_artifact_state(state):
    logger = get_run_logger()
    if state.is_failed():
        exc = state.result(raise_on_failure=False)
        logger.warning(f"Artifact failed (non-blocking): {exc}")
        return None
    return state.result()

@flow
def main():
    art_state = publish_artifact(return_state=True)  # returns a State
    handle_artifact_state(allow_failure(art_state))
3) Make the task “safe” by swallowing exceptions - If you never want the artifact task to fail (so it doesn’t show as Failed in the UI), catch exceptions inside the task and return a sentinel value:
Copy code
from prefect import flow, task, get_run_logger
from prefect.artifacts import create_markdown_artifact

@task(retries=3, retry_delay_seconds=10)
def try_publish_artifact(text: str):
    logger = get_run_logger()
    try:
        return {"artifact_id": create_markdown_artifact(markdown=text, key="daily-report")}
    except Exception as e:
        logger.warning(f"Artifact creation skipped: {e}")
        return {"artifact_id": None, "error": str(e)}  # Task is Completed, not Failed

@flow
def main():
    info = try_publish_artifact("# Hello")
    # continue regardless; info["artifact_id"] may be None
When to choose which - Use allow_failure (Pattern 1/2) if you want the artifact step to visibly fail (for observability) but not block downstream work. - Use try/except inside the task (Pattern 3) only if you truly want to avoid a Failed state for that task. Extras
- Consider adding `retries`/`retry_delay_seconds` to the artifact task for transient outages. - 
allow_failure
also works with collections and mapped tasks, so you can pass lists/dicts of futures and proceed even if some elements failed. If you share how you’d like these failures to appear in the UI (visible as Failed vs. silently handled), I can tailor the pattern and snippet to your exact use case.
22 Views
Open in Slack
PreviousNext
Powered by