How to set CPU and memory size for a Ray job?

[hongbo-miao]

My Ray cluster has different size of nodes.

Here is my current code:

import daft
import ray
from daft import DataCatalogType
from daft.io.catalog import DataCatalogTable


def hm_daft_analysis() -> None:
    ray_address = "ray://hm-ray-cluster-head-svc.production-hm-ray-cluster.svc:10001"
    ray.init(
        ray_address,
        runtime_env={
            "pip": [
                "getdaft[aws,deltalake,ray]==0.4.6",
            ],
        },
    )
    daft.context.set_runner_ray(ray_address)

    table = DataCatalogTable(
        catalog=DataCatalogType.GLUE,
        database_name="production_motor_db",
        table_name="motor_data",
    )

    df = daft.read_deltalake(table)

    # Here do some computing
    # ...
    df = df.collect()
    print(f"{df = }")


if __name__ == "__main__":
    hm_daft_analysis()

I originally thought I can set memory and CPU by

    ray.init(
        ray_address,
        runtime_env={
            "pip": [
                "getdaft[aws,deltalake,ray]==0.4.6",
            ],
        },
        num_cpus=10,
        object_store_memory=100 * 10**9,  # 100 GB
    )

However, this gives me error

Call to ray.init() on the server failed with: When connecting to an existing cluster, num_cpus and num_gpus must not be provided.

In a standard Ray job, I can set

@ray.remote(num_cpus=10, memory=100 * 10**9)

How to set How to set CPU and memory size for a Ray job when use with Daft? Thanks!

[kevinzwang]

Hi @hongbo-miao , Daft will attempt to use all of the resources available on the cluster by default, so no need to set those parameters.

If you are running a UDF, then take a look at our UDF docs to take a look at how to set the CPU and memory requirements for the UDF specifically.

For reference, here are the relevant Ray docs for ray.init(): https://docs.ray.io/en/latest/ray-core/starting-ray.html

[kevinzwang]

Daft does try to schedule some amount more resources than is immediately available in the cluster in order to trigger the Ray autoscaler. I would recommend trying it out first and see if it scales properly.

You can also configure it with the max_task_backlog parameter in daft.context.set_runner_ray, something like this:

import daft.context
daft.context.set_runner_ray(max_task_backlog=256)

[hongbo-miao]

Hi @kevinzwang for

Daft does try to schedule some amount more resources than is immediately available in the cluster in order to trigger the Ray autoscaler.

Hmm could you help explain how it works? In my case, as the scaling policy table shows, I only have 1 GB, 2 GB, 4 GB, 8 GB Ray workers running all time. For other larger Ray worker such as 16 GB, 28 GB, 58 GB, 460 GB memory, the Ray code needs to request the resource in advance to start to run it.

In traditional Ray code, when I write @ray.remote(num_cpus=10, memory=400 * 10**9), based on my scaling policy, it will start a Ray worker with 460 GB memory and 57 CPU cores.

How does Daft know how much memory the script needs before it runs in advance?
In my case, will it request

• 8 GB memory + 6 cores CPU which is largest Ray worker currently running
• or 460 GB memory + 57 cores CPU that is max resource I allow based on my scaling policy?

Also, I checked max_task_backlog at https://www.getdaft.io/projects/docs/en/latest/api_docs/doc_gen/configuration_functions/daft.context.set_runner_ray.html
It is

Maximum number of tasks that can be queued. None means Daft will automatically determine a good default.

which I feel it is for how many Ray worker Daft can use which may not related with how big each Ray worker is (?)

Thanks!

[hongbo-miao]

I found prefect-ray. The way they do is providing a remote_options:

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

@task
def process(x):
    return x + 1

@flow(
    task_runner=RayTaskRunner(
        address="ray://hm-ray-cluster-head-svc.production-hm-ray-cluster.svc:10001",
        init_kwargs={"runtime_env": {"pip": ["prefect-ray"]}},
    )
)
def my_flow():
    # equivalent to setting @ray.remote(num_cpus=4, num_gpus=2)
    with remote_options(num_cpus=4, num_gpus=2):
        process.submit(42)

Maybe it can give Daft some ideas ☺️

[kevinzwang]

Thanks for the details.

Daft does its own @ray.remote calls for each task (a unit of compute on a partition of the data) it schedules, which by default requests for num_cpus=1. So if Daft schedules a lot of tasks on Ray, the amount which is controlled by the amount of resources currently available in the Ray cluster + max_task_backlog, Ray will try to scale to meet that demand.

To my understanding what you mean is that Ray has several options on what type of machine to spin up. I'm not sure what behavior it will have with Daft (whether it will choose to spin up many small machines or a few big ones). I would encourage you to try it out and see if there are any bottlenecks and issues with the way Daft and Ray interact given your autoscaling setup.

With the current Ray runner, it doesn't make sense for Daft to ask for more than 1 CPU per task because tasks are single-threaded. We also do not account for memory usage for non-UDF scheduling purposes, although that is an area we could explore. However, we are working on an actor-based multi-threaded runner based off our new push-based local runner, where the actor resources will be an interesting factor, and maybe we can add configurations to that.

[kevinzwang]

Are there Daft workloads where you are finding that cluster machines are OOMing or where memory is a bottleneck?