Great question. For Prefect 3.x, the preferred pattern depends on whether you want to stage to local disk or stream directly to S3. Here are two solid options. Option A — Stage locally, then recursively upload the Zarr directory - Easiest to reason about, works with any Zarr directory on disk - Uses the

S3Bucket

block’s

upload_from_folder

, which preserves the directory structure

from pathlib import Path
import tempfile
import zarr
import numpy as np

from prefect import flow, task
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket


@task
def create_or_download_zarr(local_dir: Path) -> Path:
    # Example: create a small Zarr store; replace this with your actual download logic
    store_path = local_dir / "my_dataset.zarr"
    root = zarr.open_group(str(store_path), mode="w")
    root.create_dataset("data", data=np.arange(100).reshape(10, 10))
    return store_path


@flow
def upload_zarr_to_s3(bucket_name: str, s3_prefix: str = "datasets/my_zarr"):
    # Load AWS credentials (or rely on IAM/ENV if you prefer)
    aws = AwsCredentials.load("my-aws-credentials")

    s3 = S3Bucket(
        bucket_name=bucket_name,
        credentials=aws,
        # Optional: a base prefix for this bucket
        bucket_folder=None,
    )

    with tempfile.TemporaryDirectory() as tmpdir:
        local_zarr_path = create_or_download_zarr(Path(tmpdir))
        # Recursively upload all files under the Zarr directory
        s3_path = s3.upload_from_folder(
            from_folder=local_zarr_path,
            to_folder=s3_prefix,  # e.g., <s3://bucket/datasets/my_zarr/>
        )
    return s3_path


if __name__ == "__main__":
    upload_zarr_to_s3(bucket_name="my-bucket", s3_prefix="datasets/my_zarr_v1")

- S3 docs for Prefect AWS: prefect-aws S3 docs - Blocks overview: Blocks docs Option B — Stream directly between stores (no local disk) - Ideal if your source is remote (HTTP/S3/etc.) and you want to avoid local staging - Uses

fsspec

mappers and Zarr’s store-to-store copy

from prefect import flow, task
import fsspec
import zarr
from prefect_aws import AwsCredentials


@task
def copy_zarr_store(src_store_url: str, dst_s3_url: str, use_prefect_creds: bool = True) -> str:
    # src_store_url examples:
    # - "<s3://source-bucket/path/to/store.zarr>"
    # - "<https://host/path/to/store.zarr>" (must support listing)
    # dst_s3_url example:
    # - "<s3://my-bucket/datasets/my_zarr_v1/>"

    # Source mapper
    src = fsspec.get_mapper(src_store_url)

    # Destination mapper with credentials if needed
    if use_prefect_creds:
        aws = AwsCredentials.load("my-aws-credentials")
        s3_kwargs = {}
        if aws.aws_access_key_id and aws.aws_secret_access_key:
            s3_kwargs.update(
                {
                    "key": aws.aws_access_key_id,
                    "secret": aws.aws_secret_access_key,
                    "token": aws.aws_session_token,
                }
            )
        dst = fsspec.get_mapper(dst_s3_url, **s3_kwargs)
    else:
        # Relies on IAM role or environment variables
        dst = fsspec.get_mapper(dst_s3_url)

    # Copies all keys from src to dst, preserving structure
    zarr.copy_store(src, dst)
    return dst_s3_url


@flow
def mirror_zarr_to_s3(src_store_url: str, dst_s3_url: str) -> str:
    return copy_zarr_store.submit(src_store_url, dst_s3_url).result()


if __name__ == "__main__":
    mirror_zarr_to_s3(
        src_store_url="<https://example.com/path/to/store.zarr>",
        dst_s3_url="<s3://my-bucket/datasets/my_zarr_v1/>",
    )

Notes and tips - Install extras:

pip install "prefect[aws]" zarr fsspec s3fs

-

S3Bucket.upload_from_folder

is the purpose-built method for recursive uploads of directory-like datasets (like Zarr). It preserves subdirectories and filenames.

- If your source is HTTP, ensure the server supports directory listing for fsspec to map the store correctly. If not, use Option A. - You can use async flows and

await s3.aupload_from_folder(...)

for higher throughput if desired. Would you prefer local staging (Option A) or streaming (Option B)? Also, where is your Zarr source coming from (local disk, HTTP, another S3 bucket)? I can tailor the flow to your exact setup.

To run your flow in a short‑lived container that’s triggered by the Prefect server, use a containerized work pool (Docker or Kubernetes). Each flow run will spin up a fresh container, run, then exit. Here’s a minimal, reliable setup with Docker (Kubernetes is similar if you prefer): 1) Write your flow (downloads/writes Zarr, uploads to S3)

from pathlib import Path
import tempfile
import zarr
import numpy as np

from prefect import flow, task
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket


@task
def create_or_download_zarr(local_dir: Path) -> Path:
    # Replace with your actual logic to produce a local Zarr store
    store_path = local_dir / "my_dataset.zarr"
    root = zarr.open_group(str(store_path), mode="w")
    root.create_dataset("data", data=np.arange(100).reshape(10, 10))
    return store_path


@flow
def upload_zarr_to_s3(bucket_name: str, s3_prefix: str = "datasets/my_zarr"):
    aws = AwsCredentials.load("my-aws-credentials")  # create in UI or code once
    s3 = S3Bucket(bucket_name=bucket_name, credentials=aws)

    with tempfile.TemporaryDirectory() as tmpdir:
        zarr_path = create_or_download_zarr(Path(tmpdir))
        s3.upload_from_folder(from_folder=zarr_path, to_folder=s3_prefix)

2) Build a Docker image with your code and deps Dockerfile

FROM python:3.11-slim

# System deps if needed (e.g., for numpy)
RUN apt-get update && apt-get install -y build-essential && rm -rf /var/lib/apt/lists/*

# Install app deps
RUN pip install --no-cache-dir "prefect[aws]" zarr fsspec s3fs numpy

# Copy your code into the image
WORKDIR /app
COPY . /app

# Default command is not used for runs; Prefect sets it per-job

Build and push

docker build -t <your-registry>/zarr-uploader:latest .
docker push <your-registry>/zarr-uploader:latest

3) Create a Docker work pool (one-time)

prefect work-pool create "zarr-docker-pool" --type docker

4) Start a worker to poll the pool (must run on a machine with access to Docker) - Ensure the worker knows how to reach your Prefect server: - Set

PREFECT_API_URL

(e.g.,

http://<server-host>:4200/api

) and, if using Cloud,

PREFECT_API_KEY

.

export PREFECT_API_URL="http://<your-prefect-server>:4200/api"
prefect worker start --pool "zarr-docker-pool"

5) Create a deployment that uses your image and pool Option A (Python API, simplest and explicit) Create a one-off script to register the deployment:

# deploy.py
from flows.zarr_flow import upload_zarr_to_s3  # adjust import to your file/module

# We provide the image and skip building/pushing inside Prefect since we did it manually
upload_zarr_to_s3.deploy(
    name="zarr-to-s3",
    work_pool_name="zarr-docker-pool",
    image="<your-registry>/zarr-uploader:latest",
    build=False,
    push=False,
    parameters={"bucket_name": "my-bucket", "s3_prefix": "datasets/my_zarr_v1"}
)

Run once:

python deploy.py

Option B (CLI + prefect.yaml) You can also use

prefect deploy

from a project with a

prefect.yaml

that defines

image

and

work_pool

. If you want this route, I can help you scaffold a minimal

prefect.yaml

. 6) Trigger runs - From the UI (Deployments -> Run) - Or via API/CLI - Each run will create a short‑lived Docker container via the worker Notes - Make sure the container has AWS access: - Use an

AwsCredentials

block (referenced by name in the flow), or - Rely on IAM role attached to the host/runner, or - Provide `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY` in the job env (work pool template). - If you prefer Kubernetes, create a Kubernetes work pool instead, start a k8s worker, and set

image

the same way. The pattern is identical; only the infrastructure type changes. Want me to tailor this to Kubernetes, or help you with a

prefect.yaml

so you can deploy with

prefect deploy

?

Short answer: yes—run the deployment registration in CI, right after you build and push your image. It’s idempotent, so you can safely run it on every merge to main or only on releases. A typical CI order of operations - Checkout code - Build and push your container image (tag with the commit SHA or a release tag) - Set Prefect API access for CI: - Self-hosted:

PREFECT_API_URL=http://<server>:4200/api

- Cloud:

PREFECT_API_URL

(from workspace) +

PREFECT_API_KEY

- Run

deploy.py

to (re)register the deployment pointing at the new image Example: GitHub Actions snippet

name: deploy
on:
  push:
    branches: [ main ]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install deps
        run: pip install "prefect[aws]" zarr fsspec s3fs

      - name: Docker login
        run: echo "$CR_PAT" | docker login <http://ghcr.io|ghcr.io> -u "$GITHUB_ACTOR" --password-stdin
        env:
          CR_PAT: ${{ secrets.CR_PAT }}

      - name: Build and push image
        run: |
          IMAGE=<http://ghcr.io/yourorg/zarr-uploader:$|ghcr.io/yourorg/zarr-uploader:$>{{ github.sha }}
          docker build -t $IMAGE .
          docker push $IMAGE
          echo "IMAGE=$IMAGE" >> $GITHUB_ENV

      - name: Register deployment
        env:
          PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
          PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }}  # if using Cloud
        run: |
          python - <<'PY'
          import os
          from flows.zarr_flow import upload_zarr_to_s3
          upload_zarr_to_s3.deploy(
              name="zarr-to-s3",
              work_pool_name="zarr-docker-pool",
              image=os.environ["IMAGE"],
              build=False,
              push=False,
              parameters={"bucket_name": "my-bucket", "s3_prefix": "datasets/my_zarr_v1"}
          )
          PY

Notes - Do not start a worker in CI. The worker should run continuously on your infra (it pulls runs from the work pool and spins up short‑lived containers). - Run the “create work pool” step once manually (or as an infra provisioning step), not on every CI run: -

prefect work-pool create "zarr-docker-pool" --type docker

- If you prefer Git-based deployments, register in CI with

Flow.from_source(...).deploy(...)

after pushing code; the container image still controls runtime environment. Helpful docs - Deployments overview: Prefect deployments - Work pools and workers: Work pools and workers - S3 utilities used by your flow: prefect-aws S3 docs If you tell me which CI you’re on (GitHub, GitLab, etc.), I can tailor an exact pipeline for it.