Great question — you’re not imagining it. A RemoteFileSystem block uses fsspec under the hood and does not cache file contents by default, so repeated downloads can be slow. Here are practical ways to speed things up and avoid re-downloading unchanged files: 1) Add a local cache layer with fsspec (recommended) - Wrap your reads in fsspec’s SimpleCache or FileCache and store the cache on a persistent volume. This avoids re-downloading files unless they change.

import fsspec

# Example: cache S3 reads locally in /opt/prefect_cache
fs = fsspec.filesystem(
    "simplecache",
    target_protocol="s3",
    cache_storage="/opt/prefect_cache",  # persistent volume is best
    target_options={"client_kwargs": {"region_name": "us-east-1"}},
)

with fs.open("<s3://my-bucket/path/to/file.parquet>", "rb") as f:
    data = f.read()

- You can keep using your Prefect flow/tasks as-is, just switch the hot-path file reads to this fsspec FS so the cache is respected. - Docs: fsspec caching overview 2) Sync once to local storage, then read locally - On first run, sync the remote dir to a local folder (on a persistent disk). Subsequent steps read from local storage via LocalFileSystem.

from pathlib import Path
from prefect.filesystems import RemoteFileSystem, LocalFileSystem

remote = RemoteFileSystem(basepath="<s3://my-bucket/data/>")
local = LocalFileSystem(basepath="/opt/prefect_cache/data")

def read_cached(path: str) -> bytes:
    local_path = Path("/opt/prefect_cache/data") / path

    # If we have a local copy, use it
    if local_path.exists():
        return local.read_path(path)

    # Otherwise download once and cache
    content = remote.read_path(path)
    local.write_path(path, content)
    return content

- If you need to detect changes, compare remote metadata (size/mtime or provider-specific etags) before re-downloading:

info = <http://remote.filesystem.info|remote.filesystem.info>("<s3://my-bucket/data/file.parquet>")
size = info.get("size")
mtime = info.get("mtime")  # provider dependent

3) Speed up listings (won’t cache file contents) - For S3 specifically, setting

use_listings_cache: True

in the RemoteFileSystem

settings

can speed directory listings, but it does not cache file bodies.

from prefect.filesystems import RemoteFileSystem

remote = RemoteFileSystem(
    basepath="<s3://my-bucket/data/>",
    settings={"use_listings_cache": True}
)

4) Skip the download task entirely when inputs haven’t changed (Prefect task caching) - Prefect caching avoids re-running a task and re-fetching its result when inputs are unchanged and the cache is valid. This doesn’t cache file bytes itself, but it can prevent repeated downloads across runs if your download task returns the data or a local path and you persist results. - Docs: Task caching in Prefect 3 - Results storage: Results & storage Notes and best practices - RemoteFileSystem itself does not implement file-content caching; fsspec caching or a local sync strategy are the usual solutions. - For SimpleCache/FileCache to be effective across runs, point

cache_storage

at a persistent volume (Docker/K8s volume, local disk path, etc.). - Reuse the same filesystem instance within a run to benefit from connection pooling. - If you’re pulling deployment code from remote storage every run, consider baking code into your image or syncing to a persistent volume once per worker instead of per run. Helpful references - RemoteFileSystem block (API): Prefect 3 docs (see Blocks Filesystems)> - Prefect results & storage: Results and storage - Prefect task caching: Caching - fsspec caching: Caching overview If you can share:

- Which backend you’re using (S3, GCS, Azure, SMB, etc.) - Whether you have a persistent disk/volume available on your workers I can suggest the most efficient configuration (e.g., exact fsspec options or a small helper tailored to that provider).

Yes. You can bake the “sync to local, then read locally” pattern into your deployment and have workers reuse the cache across runs. The key is to: - Sync your remote data into a stable path inside the run container/host (e.g. /opt/prefect/cache) - Mount a persistent volume at that path for your worker so it survives across runs - Point your flow to read from the local cache path Below are two concrete ways to do it. Option A: prefect.yaml with pull steps (works for any storage) - Use a pull step to sync the remote directory to a persistent cache path - Set the working directory so your code can read local files - Ensure the worker mounts a persistent volume at /opt/prefect/cache Example prefect.yaml (S3 with AWS CLI; adjust for GCS/Azure)

deployments:
  - name: my-deployment
    entrypoint: flows/my_flow.py:my_flow
    work_pool:
      name: my-docker-pool   # or k8s pool
      job_variables:
        # For Docker workers: mount a persistent volume at /opt/prefect/cache
        volumes:
          - "prefect-cache:/opt/prefect/cache"
    pull:
      - prefect.deployments.steps.run_shell_script:
          script: |
            set -euo pipefail
            mkdir -p /opt/prefect/cache/data
            # Sync only changes; fast no-op if nothing changed
            aws s3 sync <s3://my-bucket/prefix> /opt/prefect/cache/data --only-show-errors
      - prefect.deployments.steps.set_working_directory:
          directory: "/opt/prefect/cache/data"

Notes: - For GCS, use gsutil:

gsutil -m rsync -r <gs://bucket/prefix> /opt/prefect/cache/data

- For Azure, use azcopy:

azcopy sync '<https://account.blob.core>.../container/prefix' '/opt/prefect/cache/data'

- Make sure your worker image includes the chosen CLI and that credentials are available (env vars, workload identity, etc.) In your flow code, read from local paths:

from pathlib import Path

def load_data():
    # Working directory is already /opt/prefect/cache/data
    return (Path("bigfile.parquet")).read_bytes()

Option B: Python deployment with pull steps

from prefect import flow
from prefect.filesystems import LocalFileSystem

@flow
def my_flow():
    # Read from local cache
    cache_dir = "/opt/prefect/cache/data"
    # ... open files under cache_dir ...

flow.from_source(
    source="<https://github.com/org/repo>",
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-docker-pool",
    pull=[
        {
            "prefect.deployments.steps.run_shell_script": {
                "script": """
                set -euo pipefail
                mkdir -p /opt/prefect/cache/data
                aws s3 sync <s3://my-bucket/prefix> /opt/prefect/cache/data --only-show-errors
                """
            }
        },
        {
            "prefect.deployments.steps.set_working_directory": {
                "directory": "/opt/prefect/cache/data"
            }
        },
    ],
)

Make the cache persist across runs - Docker worker: configure the work pool to mount a persistent volume to /opt/prefect/cache (named volume or host bind mount). Example job_variables as shown above. - Kubernetes worker: add a PVC mount in the pool’s base job template so every run mounts the same PVC at /opt/prefect/cache (volumeMount + volumes). The pull step will then “sync if needed,” and subsequent runs reuse the files. Why this works - The pull step runs at the start of each flow run and uses a delta-sync (aws s3 sync/gsutil rsync/azcopy sync), so it only downloads changed files. - Because the worker mounts a persistent volume, the cached files are available to future runs on that worker pool. - Your flow reads from the local directory, avoiding repeated remote downloads. Docs to help you put this together - Deploy via Python: Deploy via Python - prefect.yaml and pull steps: prefect.yaml

- Store flow code and pre-run pulls: Store flow code - Docker worker infrastructure: Docker worker - Kubernetes worker infrastructure: Kubernetes worker If you tell me which backend (S3/GCS/Azure) and worker type (Docker/K8s/Process), I can provide an exact, ready-to-paste config for your environment.