Daft+ray in docker-compose

[sebbegg]

Hi there,

the scaling approach of daft via ray at first appears very promising to me.
Before going to kubernetes I'd thought playing with a docker-compose based setup should be a faster start.

So I wrote a little compose yaml:

services:
  app:
    image: daft-ray-compose
    build:
      context: .
    environment:
      DAFT_RUNNER_RAY: ray-head:6379
      PYTHONUNBUFFERED: "1"
    command:
      - python
      - daft_ray_test.py
    depends_on:
      - ray-head
      - ray-workers
  ray-head:
    image: daft-ray-compose
    command: [ray, start, --head, --block, --include-dashboard=true, --dashboard-host=0.0.0.0, --temp-dir=/raytemp]
    ports:
      - 8265:8265
      - 10001:10001
    volumes:
      - ray:/raytemp
  ray-workers:
    image: daft-ray-compose
    command: [ray, start, --block, --address=ray-head:6379]
    deploy:
      replicas: 2
    depends_on:
      - ray-head

volumes:
  ray:

A tiny Dockerfile:

FROM python:3.11.9-slim

RUN pip install -U "ray[default]" daft "deltalake<1.2"

WORKDIR /app
ADD daft_ray_test.py /app

and an associated script, essentially daft_ray_test.py:

ray_runner = os.environ["DAFT_RUNNER_RAY"]
print("Connecting to ray at:", ray_runner)
daft.set_runner_ray(ray_runner)

print("Connected 🎉")
df = daft.from_pylist([{"a": x, "b": 2 * x} for x in range(1000)])
print(df.agg(df["a"].sum(), df["b"].sum()).collect())

I played a lot with ray options, runner-adresses etc. but didn't get things to work.
The closest (I think) I got is the attached configuration, which produces the following error though:

At the same time the logs of head+worker look ok and the ray dashboard also shows that the worker(s) has successfully connected to the head.
Is there anything obvious I'm missing?

[sebbegg]

Any insights? 😢

[kevinzwang]

@sebbegg this looks more like a Ray configuration issue than a Daft issue. If you just run a simple Ray script without any Daft involved, does that work? For example, this example I got from the Ray getting started guide, modified to use your env var:

import ray
import os

ray_runner = os.environ["DAFT_RUNNER_RAY"]
ray.init(ray_runner)

@ray.remote
def f(x):
    return x * x

futures = [f.remote(i) for i in range(4)]
print(ray.get(futures)) # [0, 1, 4, 9]