Short answer: yes—you need to make sure the runtime image for your flow-run pods includes

prefect-aws

(and friends). The Helm chart’s

image:

you pasted is just for the worker pod; your flows run in separate Kubernetes Job pods whose image is controlled by your Work Pool base job template or per-deployment

job_variables.image

. Recommended approaches: 1) Production (recommended): build a custom runtime image - Base it on the same Prefect image you’re using:

# Dockerfile
FROM prefecthq/prefect:3-python3.11-kubernetes

# Install AWS integrations
RUN pip install --no-cache-dir "prefect[aws]" boto3

- Push it to your registry and set it as the job image: - In the Work Pool (UI): set Base job template -> image to your custom image - Or per deployment via prefect.yaml:

# prefect.yaml
deployments:
  - name: s3-flow
    entrypoint: flows/s3_flow.py:flow
    work_pool:
      name: my-k8s-pool
      job_variables:
        image: <http://ghcr.io/org/prefect-aws:2024-12-24|ghcr.io/org/prefect-aws:2024-12-24>

- Or via CLI override:

prefect deploy -n s3-flow --job-variable image=<http://ghcr.io/org/prefect-aws:2024-12-24|ghcr.io/org/prefect-aws:2024-12-24>

2) Quick/temporary: runtime pip install - Add an env var so the job container installs packages at start:

# prefect.yaml
deployments:
  - name: s3-flow
    entrypoint: flows/s3_flow.py:flow
    work_pool:
      name: my-k8s-pool
      job_variables:
        env:
          EXTRA_PIP_PACKAGES: "prefect-aws boto3"

- Pros: fast to try. Cons: slower startup and less deterministic. Use for dev, not prod. Notes and tips: - Using

prefect[aws]

is fine; it pulls in the

prefect-aws

collection for Prefect 3.x. - Installing packages into the worker image will not fix import errors in flow runs unless the job image also has them—so prefer a custom job image or the

EXTRA_PIP_PACKAGES

env on the job. - You can also set

job_variables.image

directly in Python:

from prefect import flow

@flow
def my_flow(): ...

my_flow.deploy(
    name="s3-flow",
    work_pool_name="my-k8s-pool",
    job_variables={"image": "<http://ghcr.io/org/prefect-aws:2024-12-24|ghcr.io/org/prefect-aws:2024-12-24>"}
)

If you share how you’re creating deployments today (prefect.yaml +

prefect deploy

vs

.deploy()

), I can paste an exact snippet tailored to your setup.

Short version: - You can’t use different containers per task. With the Kubernetes worker, a whole flow run executes in a single pod/image. - To use both prefect-dbt (dbt + Snowflake) and prefect-aws in one flow, either: 1) Build one runtime image that includes both sets of deps, or 2) Split into multiple deployments (each with its own image) and orchestrate them with

run_deployment

. Recommended patterns A) One combined runtime image (simple, production-friendly) - Base it on the same family as your worker image:

# Dockerfile
FROM prefecthq/prefect:3-python3.11-kubernetes

# dbt + Snowflake + Prefect integrations + AWS
RUN pip install --no-cache-dir \
  "prefect[aws]" \
  "prefect-dbt[snowflake]" \
  dbt-snowflake \
  boto3

- Point your deployment to this image (work pool default or per-deployment):

# prefect.yaml
deployments:
  - name: etl-dbt-and-aws
    entrypoint: flows/etl.py:etl_flow
    work_pool:
      name: my-k8s-pool
      job_variables:
        image: <http://ghcr.io/org/prefect-etl:2024-12-24|ghcr.io/org/prefect-etl:2024-12-24>
        # if using IRSA, set the SA the pod must use:
        service_account_name: prefect-flow-sa

B) Split images per concern and orchestrate (flexible, “per-task-like”) - Create two deployments, one with a dbt+Snowflake image, one with an AWS image; then call them from an orchestrator flow:

from prefect import flow
from prefect.deployments import run_deployment

@flow
async def pipeline_orchestrator():
    await run_deployment("dbt/dbt-snowflake-deployment", parameters={"select": "tag:daily"})
    await run_deployment("aws/aws-tasks-deployment", parameters={"bucket": "my-bucket"})

dbt + Snowflake specifics - dbt uses a profiles.yml. Common options: - Bake a non-secret template into your image and supply secrets via env vars. - Mount a Kubernetes Secret as

/root/.dbt/profiles.yml

(and set

DBT_PROFILES_DIR

). - Example profiles.yml using env vars (safe pattern):

# profiles.yml
my_project:
  target: prod
  outputs:
    prod:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      password: "{{ env_var('SNOWFLAKE_PASSWORD') }}"
      role: "{{ env_var('SNOWFLAKE_ROLE') }}"
      warehouse: "{{ env_var('SNOWFLAKE_WAREHOUSE') }}"
      database: "{{ env_var('SNOWFLAKE_DATABASE') }}"
      schema: "{{ env_var('SNOWFLAKE_SCHEMA') }}"
      threads: 4

- In your deployment, set the env and (optionally) `DBT_PROFILES_DIR`:

work_pool:
  name: my-k8s-pool
  job_variables:
    image: <http://ghcr.io/org/prefect-etl:2024-12-24|ghcr.io/org/prefect-etl:2024-12-24>
    env:
      DBT_PROFILES_DIR: /root/.dbt
      SNOWFLAKE_ACCOUNT: xxxxxx
      SNOWFLAKE_USER: svc_user
      SNOWFLAKE_PASSWORD: ${SNOWFLAKE_PASSWORD}  # from K8s secret/Env
      SNOWFLAKE_ROLE: TRANSFORMER
      SNOWFLAKE_WAREHOUSE: WH_XS
      SNOWFLAKE_DATABASE: ANALYTICS
      SNOWFLAKE_SCHEMA: TRANSFORM

- Minimal flow using prefect-dbt:

from prefect import flow
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings

@flow(log_prints=True)
def dbt_flow(select: str = "tag:daily"):
    settings = PrefectDbtSettings(
        project_dir="/opt/prefect/flows/my_dbt_project",  # copy or clone repo into image
        profiles_dir="/root/.dbt"
    )
    PrefectDbtRunner(settings=settings).invoke(["run", "--select", select])

AWS access inside the same flow - With the combined image, just import and use prefect-aws/boto3 tasks directly:

from prefect import flow
from prefect_aws import AwsCredentials
from prefect_aws.s3 import s3_download

@flow
def aws_flow(bucket: str, key: str):
    # If using IRSA, you can often omit keys; boto3 picks up the pod role
    creds = AwsCredentials(region_name="us-east-1")
    data = s3_download(bucket=bucket, key=key, aws_credentials=creds)
    return data

- If using IRSA, ensure the flow-run pod uses an annotated ServiceAccount: - In job variables:

service_account_name: <sa-with-iam-role>

- Annotate that SA with the AWS role that permits S3/Secrets Manager/etc. When to choose which approach - Single combined image: - Best if dbt + AWS are frequently used together. - Simplest operationally; fastest runtime (no pip at boot). - Separate deployments + run_deployment: - Best if dependencies diverge a lot or you want independent scaling/retries. - Mimics “per-task image” by splitting work across flow runs. If you share how you’re building deployments today (prefect.yaml vs

.deploy()

), I can drop in an exact snippet for your setup, including profiles.yml handling and IRSA-friendly settings.