job_variables

from prefect.deployments import run_deployment

run_deployment(
    name="my-flow/my-deployment",
    parameters={"some_param": "value"},
    job_variables={"image": "<http://ghcr.io/your-org/prefect-ray:py310-cu121|ghcr.io/your-org/prefect-ray:py310-cu121>"},
)

# sketch
my_flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="flows/ray_flow.py:my_flow",
).deploy(
    name="ray-flow",
    work_pool_name="kubernetes-pool",  # or docker, ecs, etc.
    image="<http://ghcr.io/your-org/prefect-ray:py310|ghcr.io/your-org/prefect-ray:py310>",
)

RayTaskRunner(address=...)

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(task_runner=RayTaskRunner(address="<ray://your-ray-head:10001>"))
def my_flow():
    # your tasks here; they will run on the Ray cluster
    ...

# Deploy this flow with different images per deployment or override per run via job_variables

init_kwargs

RayTaskRunner

address="ray://..."

init_kwargs={"runtime_env": {"image_uri": ...}}

init_kwargs={"runtime_env": {"container": {"image": ...}}}

RayTaskRunner(address=...)

address

init_kwargs={"runtime_env": {"container": {"image": "your-image"}}}

image_uri

RayTaskRunner

ray.init(...)

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

# Connect to an existing cluster (won’t change its pod image)
@flow(task_runner=RayTaskRunner(address="<ray://ray-head:10001>",
                                init_kwargs={"runtime_env": {"env_vars": {"FOO": "bar"}}}))
def my_flow():
    ...

# Start a local Ray runtime and run tasks in a Docker image (host must have Docker)
@flow(task_runner=RayTaskRunner(init_kwargs={
    "runtime_env": {"container": {"image": "<http://ghcr.io/your-org/ray-runtime:py310|ghcr.io/your-org/ray-runtime:py310>"}}
}))
def local_containerized_flow():
    ...

from prefect.deployments import run_deployment

run_deployment(
    name="my-flow/my-deployment",
    job_variables={"image": "<http://ghcr.io/your-org/prefect-flow:py310-cu121|ghcr.io/your-org/prefect-flow:py310-cu121>"},
)

RayTaskRunner(address=...)

init_kwargs={"runtime_env": {"image_uri": ...}}

init_kwargs={"runtime_env": {"container": {"image": ...}}}

ray.init(...)

docker

RayTaskRunner(address=...)

# flow.py
import os
from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

RAY_ADDRESS = os.getenv("RAY_ADDRESS")  # e.g., ray://<head-ip>:10001

@flow(task_runner=RayTaskRunner(address=RAY_ADDRESS))
def my_flow():
    ...

# When deploying, set the env var per deployment/run
# e.g., via job_variables
# job_variables={"env": {"RAY_ADDRESS": "<ray://ec2-gpu-head:10001>"}}

runtime_env={"image_uri": "<your-image>"}

from ray.job_submission import JobSubmissionClient

client = JobSubmissionClient("http://<head-node-ip>:8265")
job_id = client.submit_job(
    entrypoint="python -c 'import your_module; your_module.main()'",
    runtime_env={"image_uri": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/ray:gpu|123456789012.dkr.ecr.us-east-1.amazonaws.com/ray:gpu>"},
    # env_vars / working_dir / pip, etc. as needed
)

RayTaskRunner

ray.init(...)

@flow(task_runner=RayTaskRunner(
    address="ray://<head-ip>:10001",
    init_kwargs={"runtime_env": {"container": {"image": "your-image:tag"}}},
))
def my_flow():
    ...

address

from prefect.deployments import run_deployment

run_deployment(
    name="my-flow/my-deployment",
    job_variables={"image": "<http://ghcr.io/your-org/prefect-flow:py310-cu121|ghcr.io/your-org/prefect-flow:py310-cu121>"},
)

ray.init()

address="ray://<head-ip>:10001"

RayTaskRunner

ray.init(...)

ray.init

ray.init(runtime_env=...)

remote_options(runtime_env=...)

ray

init_kwargs={"runtime_env": {"container": {"image": "...", ...}}}

RayTaskRunner

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(
    task_runner=RayTaskRunner(
        address="ray://<head-ip>:10001",
        init_kwargs={
            "runtime_env": {
                "container": {
                    "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>",
                    # optional: pass Docker run flags
                    "run_options": [
                        "--gpus=all",           # if you need GPUs
                        "--ipc=host",
                        # "--volume=/host/path:/container/path"
                    ],
                    "env_vars": {"EXTRA_FLAG": "1"},
                }
            }
        },
    )
)
def my_flow():
    # Prefect tasks in this flow will run as Ray tasks inside the container image above
    ...

remote_options

.options(...)

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def heavy_gpu_work(x): ...
@task
def cpu_only_work(y): ...

@flow(task_runner=RayTaskRunner(address="ray://<head-ip>:10001"))
def my_flow(image_for_gpu: str, image_for_cpu: str):
    # GPU section
    with remote_options(
        runtime_env={
            "container": {
                "image": image_for_gpu,
                "run_options": ["--gpus=all"],
            }
        },
        num_gpus=1,
    ):
        a = heavy_gpu_work.submit(1)

    # CPU section
    with remote_options(
        runtime_env={"container": {"image": image_for_cpu}},
        num_cpus=2,
    ):
        b = cpu_only_work.submit(2)

    return a, b

--gpus=all

run_options

run_options

remote_options(runtime_env=...)

ray.init(runtime_env=...)

runtime_env

runtime_env

run_options

sudo apt-get update
      sudo apt-get install -y <http://docker.io|docker.io>
      sudo usermod -aG docker $USER
      # re-login or: newgrp docker
      docker info  # should succeed without sudo

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(
    task_runner=RayTaskRunner(
        address="ray://<head-ip>:10001",
        init_kwargs={
            "runtime_env": {
                "container": {
                    "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>",
                    # optional: docker run flags
                    "run_options": [
                        "--gpus=all",     # if you need GPUs
                        "--ipc=host",
                        # "--volume=/host/path:/container/path"
                    ],
                    # pass env to the container
                    "env_vars": {"EXTRA_FLAG": "1"},
                }
            }
        },
    )
)
def my_flow():
    ...

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def heavy_gpu_work(x): ...
@task
def cpu_only_work(y): ...

@flow(task_runner=RayTaskRunner(address="ray://<head-ip>:10001"))
def my_flow(image_for_gpu: str, image_for_cpu: str):
    # GPU section
    with remote_options(
        runtime_env={"container": {"image": image_for_gpu, "run_options": ["--gpus=all"]}},
        num_gpus=1,
    ):
        a = heavy_gpu_work.submit(1)

    # CPU section
    with remote_options(
        runtime_env={"container": {"image": image_for_cpu}},
        num_cpus=2,
    ):
        b = cpu_only_work.submit(2)

    return a, b

ray.init()

ray://...

ray.init(...)

address="ray://<head-ip>:10001"

--gpus=all

docker pull <your-image>

ray.init(address="ray://...")

runtime_env={"container": {"image": "<your-image>"}}

echo $PATH
which docker

which docker

sudo yum install -y docker
sudo systemctl enable docker
sudo systemctl start docker
sudo usermod -aG docker ec2-user

ec2-user

docker

newgrp docker

groups

docker

which docker

/usr/bin

docker

which docker
docker ps
groups

docker

docker ps

docker

newgrp docker

What to look for:
- The “Without container” and “With container” sections should differ in `os_release` (confirming your container image ran).
- If you get errors like “podman not found” or “docker: permission denied”:
  - Make sure `which docker` returns a path in both the driver and inside the non-container task.
  - Re-check group membership took effect (new login/newgrp).
  - Ensure Docker is installed on every worker node, not just the head.
  - Try pre-pulling the image on nodes (`docker pull <IMAGE>`) to rule out registry/auth issues.

Using this with Prefect + Ray
Once the node-level test works, apply the same runtime_env via Prefect-Ray:

- Default container for all tasks in a flow:

- Different images per section/task:

docker

ray attach ray-cluster-config-docker.yaml

which docker
docker ps
id -nG

docker ps

docker

newgrp docker
docker ps

which docker

/usr/bin

export PATH="/usr/bin:$PATH"
which docker

docker pull <http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>

ray status
python -c "import ray; print(ray.__version__)"

python3

python

python - << 'PY'
import os, shutil, ray, subprocess, json

IMAGE = "123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310"

print("Driver PATH:", os.environ.get("PATH"))
print("Driver which(docker):", shutil.which("docker"))

# Connect to the local running cluster on the head
ray.init(address="auto")

@ray.remote
def info():
    import os, subprocess, shutil
    return {
        "PATH": os.environ.get("PATH"),
        "which_docker": shutil.which("docker"),
        "whoami": subprocess.getoutput("whoami"),
        "python": subprocess.getoutput("python -V 2>&1"),
        "os_release": subprocess.getoutput("cat /etc/os-release || true"),
        "has_nvidia_smi": subprocess.call("nvidia-smi >/dev/null 2>&1", shell=True) == 0,
    }

print("\n--- Without container ---")
print(json.dumps(ray.get(info.remote()), indent=2))

print("\n--- With container runtime_env ---")
task = info.options(
    runtime_env={
        "container": {
            "image": IMAGE,
            # Uncomment/add if needed:
            # "run_options": ["--gpus=all", "--ipc=host"],
            # "env_vars": {"EXAMPLE": "1"},
        }
    }
).remote()

print(json.dumps(ray.get(task), indent=2))
PY

os_release

which_docker

/usr/bin

docker

newgrp docker

ray stop

ray up ray-cluster-config-docker.yaml

ray up

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(task_runner=RayTaskRunner(
    address="ray://<head-ip>:10001",  # or address="auto" if running on head
    init_kwargs={
        "runtime_env": {
            "container": {
                "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310|123456789012.dkr.ecr.us-east-1.amazonaws.com/your-image:py310>",
                # "run_options": ["--gpus=all"]
            }
        }
    }
))
def my_flow():
    ...

from prefect_ray.context import remote_options
with remote_options(runtime_env={"container": {"image": "your-other-image"}}):
    ...

which docker

docker ps

id -nG

ssh -i /path/to/your/key.pem ec2-user@172.31.25.199

which docker
docker ps
id -nG

sudo yum install -y docker
sudo systemctl enable --now docker
sudo usermod -aG docker ec2-user
# Open a fresh login OR run:
newgrp docker
docker ps

# ray-cluster-config-docker.yaml (excerpt)
auth:
  ssh_user: ec2-user

setup_commands:
  - sudo yum install -y docker
  - sudo systemctl enable --now docker
  - sudo usermod -aG docker ec2-user
  # ensure current shell has docker group for subsequent commands
  - newgrp docker || true
  - docker --version

ray up

RayTaskRunner(init_kwargs={...})

docker ps | grep -i ray

ps -ef | grep raylet

sudo nsenter -t <raylet-pid> -m -u -i -n env | grep PATH

docker exec -it <ray-container-name> bash -lc 'which docker || echo no-docker; ls -l /var/run/docker.sock || echo no-sock'

FROM rayproject/ray:2.48.0-py310
      RUN apt-get update && apt-get install -y <http://docker.io|docker.io> && rm -rf /var/lib/apt/lists/*

--volume=/var/run/docker.sock:/var/run/docker.sock

--ipc=host

--privileged

pgrep -f raylet

sudo tr '\0' '\n' </proc/$(pgrep -f raylet | head -n1)/environ | grep PATH

# 1) Confirm raylet PID (on the worker)
pgrep -f raylet

# 2) Show the PATH visible to the raylet process (replace <PID> with output above)
sudo tr '\0' '\n' </proc/<PID>/environ | grep ^PATH=

# 3) Sanity check your current shell
echo $PATH
which docker
id -nG

# 4) Check runtime env agent logs for clues
ls -1 /tmp/ray/session_*/logs | grep runtime_env
grep -i -E 'podman|docker' /tmp/ray/session_*/logs/runtime_env* 2>/dev/null | tail -n 100

python - << 'PY'
import ray, json, os, shutil

ray.init(address="auto")

# Make an info task
@ray.remote
def info():
    import os, shutil, subprocess
    return {
        "PATH": os.environ.get("PATH"),
        "which_docker": shutil.which("docker"),
        "whoami": subprocess.getoutput("whoami"),
    }

# Force the task to run on the target worker via node resource
target_ip = "172.31.25.199"
res = ray.get(info.options(resources={f"node:{target_ip}": 0.001}).remote())
print(json.dumps(res, indent=2))
PY

which_docker

head_start_ray_commands:
      - 'export PATH=/usr/bin:$PATH; ray stop'
      - 'export PATH=/usr/bin:$PATH; ray start --head --dashboard-host=0.0.0.0 --port=6379'
    worker_start_ray_commands:
      - 'export PATH=/usr/bin:$PATH; ray stop'
      - 'export PATH=/usr/bin:$PATH; ray start --address=$RAY_HEAD_IP:6379'

ray up

groups

newgrp docker

ray start

E) Why “podman: not found” despite docker being installed
- Ray’s containerized tasks feature shells out from the raylet/runtime-env agent. If the raylet’s environment cannot find docker (binary not on PATH at the time the process was started), Ray may try podman. Your interactive shell is fine, but the raylet process environment may be missing PATH entries.
- Ensuring PATH (and group membership) for the process that starts Ray fixes this.

Once this test works, you can use the same runtime_env in Prefect-Ray’s RayTaskRunner:
- Global for the flow:

remote_options