Short answer: With Prefect 3.x, the Kubernetes worker creates a new Kubernetes Job/Pod for every flow run by design. If you want to run every 2 minutes without creating a new pod each time, run a long‑lived Process worker inside a single Kubernetes Deployment. That worker will execute each run as a subprocess in the same pod. Two ways to proceed: A) Preferred to avoid new pods: Run a Process worker in a long‑lived K8s Deployment - Create a Process work pool:

prefect work-pool create my-process-pool --type process

- Deploy your flow on a 2‑minute interval to that pool:

from prefect import flow
from datetime import timedelta

@flow
def my_flow():
    print("hello every 2 minutes")

if __name__ == "__main__":
    # If your worker image already includes this code:
    my_flow.deploy(
        name="two-mins",
        work_pool_name="my-process-pool",
        interval=timedelta(minutes=2),
    )

    # Or pull code from source:
    # from prefect import flow
    # flow.from_source(
    #   "<https://github.com/your-org/your-repo>",
    #   entrypoint="flows/my_flow.py:my_flow",
    # ).deploy(
    #   name="two-mins",
    #   work_pool_name="my-process-pool",
    #   interval=timedelta(minutes=2),
    # )

- Run the worker as a Deployment (stays up; no new pods per run):

apiVersion: apps/v1
kind: Deployment
metadata:
  name: prefect-process-worker
spec:
  replicas: 1
  selector:
    matchLabels:
      app: prefect-process-worker
  template:
    metadata:
      labels:
        app: prefect-process-worker
    spec:
      containers:
        - name: worker
          image: <http://ghcr.io/prefecthq/prefect:3-latest|ghcr.io/prefecthq/prefect:3-latest>  # or your custom image with flow deps
          env:
            # For Prefect Cloud, set PREFECT_API_URL + PREFECT_API_KEY
            # For self-hosted Prefect Server, set PREFECT_API_URL (e.g., <http://prefect-api:4200/api>)
            - name: PREFECT_API_URL
              value: "<http://prefect-api:4200/api>"
          command: ["prefect", "worker", "start", "-p", "my-process-pool"]

Notes: - Each scheduled run executes as a subprocess inside this single pod. - Scale replicas >1 to spread runs across a fixed number of long‑lived pods (still no per‑run pods). - Tradeoff: less isolation; the pod becomes a single point of failure unless you run multiple replicas. B) If you stick with the Kubernetes worker (new pod per run), minimize overhead - This is the standard, recommended isolation model. For a 2‑minute schedule it’s usually fine. - Optimize startup: - Pre‑pull images on nodes; keep images small. - Use IfNotPresent image pull policy in your pool’s base job template. - Set ttlSecondsAfterFinished to clean up jobs/pods quickly. - You can set these via a custom base job template on the Kubernetes work pool:

prefect work-pool create my-k8s-pool --type kubernetes --base-job-template ./base-job-template.json

Where your base-job-template.json sets fields like ttlSecondsAfterFinished and imagePullPolicy. Key confirmations and references - Flow.deploy supports interval/cron/rrule and schedules list: - Flow.deploy signature in source: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/flows.py - Schedules (Interval, Cron, RRule) overview: https://docs.prefect.io/latest/concepts/schedules/ - Work pools and workers concepts: https://docs.prefect.io/latest/concepts/work-pools/ - Kubernetes infrastructure guide: https://docs.prefect.io/latest/deploy/infrastructure/kubernetes/ Would you like a ready-to-apply Kubernetes manifest for the Process worker using your Prefect API config, or an example base job template for the Kubernetes worker path?