@task(task_config=RayJobConfig(...), requests=Resources(cpu="2"), limits=Resources(gpu="2"))

task_resource_defaults.yaml: |
    task_resources:
      defaults:
        cpu: 400m
        memory: 500Mi
        storage: 500Mi
      limits:
        cpu: 2
        gpu: 1
        memory: 4Gi
        storage: 20Mi

cluster_resources:
      customData:
      - production:
        - projectQuotaCpu:
            value: "5"
        - projectQuotaMemory:
            value: 4000Mi
      - staging:
        - projectQuotaCpu:
            value: "2"
        - projectQuotaMemory:
            value: 3000Mi
      - development:
        - projectQuotaCpu:
            value: "12"
        - projectQuotaMemory:
            value: 8000Mi

import typing

import ray
from ray import tune
from flytekit import Resources, task, workflow
from flytekitplugins.ray import HeadNodeConfig, RayJobConfig, WorkerNodeConfig


@ray.remote
def objective(config):
    return (config["x"] * config["x"])


ray_config = RayJobConfig(
    head_node_config=HeadNodeConfig(ray_start_params={"log-color": "True"}),
    worker_node_config=[WorkerNodeConfig(group_name="ray-group", replicas=2)],
    runtime_env={"pip": ["numpy", "pandas"]},
)


@task(task_config=ray_config, limits=Resources(mem="2000Mi", cpu="1"))
def ray_task(n: int) -> int:
    model_params = {
        "x": tune.randint(-10, 10)
    }

    tuner = tune.Tuner(
        objective,
        tune_config=tune.TuneConfig(
            num_samples=10,
            max_concurrent_trials=n,
        ),

        param_space=model_params,
    )
    results = tuner.fit()
    return results


@workflow
def ray_workflow(n: int) -> int:
    return ray_task(n=n)

Placement group creation timed out. Make sure your cluster either has enough resources or use an autoscaling cluster. Current resources available: {'CPU': 1.0, 'object_store_memory': 217143276.0, 'memory': 434412750.0, 'node:10.69.53.118': 0.98}, resources requested by the placement group: [{'CPU': 1.0}, {'CPU': 1.0}]

from flytekitplugins.ray import HeadNodeConfig, RayJobConfig, WorkerNodeConfig
import ray
from ray import tune

#ray.init()
#ray.init("auto", ignore_reinit_error=True)

ray_config = RayJobConfig(
    head_node_config=HeadNodeConfig(ray_start_params={"log-color": "True"}),
    worker_node_config=[WorkerNodeConfig(group_name="ray-group", replicas=2)],
)

num_actors = 4
num_cpus_per_actor = 1

ray_params = RayParams(
    num_actors=num_actors, cpus_per_actor=num_cpus_per_actor)


def train_model(config):
    train_x, train_y = load_breast_cancer(return_X_y=True)
    train_set = RayDMatrix(train_x, train_y)

    evals_result = {}
    bst = train(
        params=config,
        dtrain=train_set,
        evals_result=evals_result,
        evals=[(train_set, "train")],
        verbose_eval=False,
        ray_params=ray_params)
    bst.save_model("model.xgb")



@task(task_config=ray_config, limits=Resources(mem="2000Mi", cpu="1"))
def train_model_task() -> dict:
    config = {
        "tree_method": "approx",
        "objective": "binary:logistic",
        "eval_metric": ["logloss", "error"],
        "eta": tune.loguniform(1e-4, 1e-1),
        "subsample": tune.uniform(0.5, 1.0),
        "max_depth": tune.randint(1, 9)
    }


    analysis = tune.run(
        train_model,
        config=config,
        metric="train-error",
        mode="min",
        num_samples=4,
        resources_per_trial=ray_params.get_tune_resources())
    return analysis.best_config

@workflow
def train_model_wf() -> dict:
    return train_model_task()

ttlSecondsAfterFinished

$ k describe rayjobs feb5da8c2a2394fb4ac8-n0-0 -n flytesnacks-development
...
  Ttl Seconds After Finished:   3600

pyflyte register --image <image name> <file name> --version <version number>

pyflyte --config ~/.flyte/config-remote.yaml run --remote --image <image_name> ray_demo.py wf

pyflyte --config ~/.flyte/config-remote.yaml run --image <image_name> ray_demo.py wf

2023-02-23T18:08:48.386Z    INFO    controllers.RayJob    RayJob associated rayCluster found    {"rayjob": "a8xqtvnds2sp7bgkn96k-fzvfpg5y-0", "raycluster": "ntropy-development/a8xqtvnds2sp7bgkn96k-fzvfpg5y-0-raycluster-xlg4f"}
2023-02-23T18:08:48.387Z    INFO    controllers.RayJob    waiting for the cluster to be ready    {"rayCluster": "a8xqtvnds2sp7bgkn96k-fzvfpg5y-0-raycluster-xlg4f"}
2023-02-23T18:08:51.387Z    INFO    controllers.RayJob    reconciling RayJob    {"NamespacedName": "ntropy-development/a8xqtvnds2sp7bgkn96k-fzvfpg5y-0"}
2023-02-23T18:08:51.388Z    INFO    controllers.RayJob    RayJob associated rayCluster found    {"rayjob": "a8xqtvnds2sp7bgkn96k-fzvfpg5y-0", "raycluster": "ntropy-development/a8xqtvnds2sp7bgkn96k-fzvfpg5y-0-raycluster-xlg4f"}
2023-02-23T18:08:51.388Z    INFO    controllers.RayJob    waiting for the cluster to be ready    {"rayCluster": "a8xqtvnds2sp7bgkn96k-fzvfpg5y-0-raycluster-xlg4f"}
2023-02-23T18:08:54.388Z    INFO    controllers.RayJob    reconciling RayJob    {"NamespacedName": "ntropy-development/a8xqtvnds2sp7bgkn96k-fzvfpg5y-0"}
2023-02-23T18:08:54.388Z    INFO    controllers.RayJob    RayJob associated rayCluster found    {"rayjob": "a8xqtvnds2sp7bgkn96k-fzvfpg5y-0", "raycluster": "ntropy-development/a8xqtvnds2sp7bgkn96k-fzvfpg5y-0-raycluster-xlg4f"}
2023-02-23T18:08:54.389Z    INFO    controllers.RayJob    waiting for the cluster to be ready    {"rayCluster": "a8xqtvnds2sp7bgkn96k-fzvfpg5y-0-raycluster-xlg4f"}
2023-02-23T18:08:57.389Z    INFO    controllers.RayJob    reconciling RayJob    {"NamespacedName": "ntropy-development/a8xqtvnds2sp7bgkn96k-fzvfpg5y-0"}