Core API Reference¶

Core dagster-ray APIs for using external Ray clusters. Learn how to use it here.

Ray Resources¶

RayResource can be used to connect to external Ray clusters when provided as a Dagster resource, or as a type annotation (all other Ray resources in dagster-ray inherit from RayResource)

dagster_ray.RayResource `pydantic-model` ¶

Bases: ConfigurableResource, ABC

Base class for Ray Resources providing a common interface for Ray cluster management.

This abstract base class defines the interface that all Ray resources must implement, providing a backend-agnostic way to interact with Ray clusters. Concrete implementations include LocalRay for local development and KubeRay resources for Kubernetes deployments.

The RayResource handles the lifecycle of Ray clusters including creation, connection, and cleanup, with configurable policies for each stage.

Example

Use as a type annotation for backend-agnostic code

import dagster as dg
from dagster_ray import RayResource

@dg.asset
def my_asset(ray_cluster: RayResource):
    # Works with any Ray backend
    import ray
    return ray.get(ray.put("hello"))

Example

Manual lifecycle management

from dagster_ray import Lifecycle

ray_resource = SomeRayResource(
    lifecycle=Lifecycle(
        create=False,  # Don't auto-create
        connect=False  # Don't auto-connect
    )
)

Note

This is an abstract class and cannot be instantiated directly. Use concrete implementations like LocalRay or KubeRayCluster instead.

Show JSON schema:

{
  "$defs": {
    "ExecutionOptionsConfig": {
      "properties": {
        "cpu": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Cpu"
        },
        "gpu": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Gpu"
        },
        "object_store_memory": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Object Store Memory"
        }
      },
      "title": "ExecutionOptionsConfig",
      "type": "object"
    },
    "Lifecycle": {
      "properties": {
        "create": {
          "default": true,
          "description": "Whether to create the resource. If set to `False`, the user can manually call `.create` instead.",
          "title": "Create",
          "type": "boolean"
        },
        "wait": {
          "default": true,
          "description": "Whether to wait for the remote Ray cluster to become ready to accept connections. If set to `False`, the user can manually call `.wait` instead.",
          "title": "Wait",
          "type": "boolean"
        },
        "connect": {
          "default": true,
          "description": "Whether to run `ray.init` against the remote Ray cluster. If set to `False`, the user can manually call `.connect` instead.",
          "title": "Connect",
          "type": "boolean"
        },
        "cleanup": {
          "default": "always",
          "description": "Resource cleanup policy. Determines when the resource should be deleted after Dagster step execution or during interruption.",
          "enum": [
            "never",
            "always",
            "on_exception"
          ],
          "title": "Cleanup",
          "type": "string"
        }
      },
      "title": "Lifecycle",
      "type": "object"
    },
    "RayDataExecutionOptions": {
      "properties": {
        "execution_options": {
          "$ref": "#/$defs/ExecutionOptionsConfig"
        },
        "cpu_limit": {
          "default": 5000,
          "title": "Cpu Limit",
          "type": "integer"
        },
        "gpu_limit": {
          "default": 0,
          "title": "Gpu Limit",
          "type": "integer"
        },
        "verbose_progress": {
          "default": true,
          "title": "Verbose Progress",
          "type": "boolean"
        },
        "use_polars": {
          "default": true,
          "title": "Use Polars",
          "type": "boolean"
        }
      },
      "title": "RayDataExecutionOptions",
      "type": "object"
    }
  },
  "description": "Base class for Ray Resources providing a common interface for Ray cluster management.\n\nThis abstract base class defines the interface that all Ray resources must implement,\nproviding a backend-agnostic way to interact with Ray clusters. Concrete implementations\ninclude LocalRay for local development and KubeRay resources for Kubernetes deployments.\n\nThe RayResource handles the lifecycle of Ray clusters including creation, connection,\nand cleanup, with configurable policies for each stage.\n\nExample:\n    Use as a type annotation for backend-agnostic code\n    ```python\n    import dagster as dg\n    from dagster_ray import RayResource\n\n    @dg.asset\n    def my_asset(ray_cluster: RayResource):\n        # Works with any Ray backend\n        import ray\n        return ray.get(ray.put(\"hello\"))\n    ```\n\nExample:\n    Manual lifecycle management\n    ```python\n    from dagster_ray import Lifecycle\n\n    ray_resource = SomeRayResource(\n        lifecycle=Lifecycle(\n            create=False,  # Don't auto-create\n            connect=False  # Don't auto-connect\n        )\n    )\n    ```\n\nNote:\n    This is an abstract class and cannot be instantiated directly. Use concrete\n    implementations like LocalRay or KubeRayCluster instead.",
  "properties": {
    "lifecycle": {
      "$ref": "#/$defs/Lifecycle",
      "description": "Actions to perform during resource setup."
    },
    "timeout": {
      "default": 600.0,
      "description": "Timeout for Ray readiness in seconds",
      "title": "Timeout",
      "type": "number"
    },
    "ray_init_options": {
      "description": "Additional keyword arguments to pass to `ray.init()` call, such as `runtime_env`, `num_cpus`, etc. Dagster's `EnvVar` is supported. More details in [Ray docs](https://docs.ray.io/en/latest/ray-core/api/doc/ray.init.html).",
      "title": "Ray Init Options",
      "type": "object"
    },
    "data_execution_options": {
      "$ref": "#/$defs/RayDataExecutionOptions"
    },
    "redis_port": {
      "default": 10001,
      "description": "Redis port for connection. Make sure to match with the actual available port.",
      "title": "Redis Port",
      "type": "integer"
    },
    "dashboard_port": {
      "default": 8265,
      "description": "Dashboard port for connection. Make sure to match with the actual available port.",
      "title": "Dashboard Port",
      "type": "integer"
    },
    "env_vars": {
      "anyOf": [
        {
          "additionalProperties": {
            "type": "string"
          },
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "description": "Environment variables to pass to the Ray cluster.",
      "title": "Env Vars"
    },
    "enable_tracing": {
      "default": false,
      "description": "Enable tracing: inject `RAY_PROFILING=1` and `RAY_task_events_report_interval_ms=0` into the Ray cluster configuration. This allows using `ray.timeline()` to fetch recorded task events. Learn more: [KubeRay docs](https://docs.ray.io/en/latest/ray-core/api/doc/ray.timeline.html#ray-timeline)",
      "title": "Enable Tracing",
      "type": "boolean"
    },
    "enable_actor_task_logging": {
      "default": false,
      "description": "Enable actor task logging: inject `RAY_ENABLE_RECORD_ACTOR_TASK_LOGGING=1` into the Ray cluster configuration.",
      "title": "Enable Actor Task Logging",
      "type": "boolean"
    },
    "enable_debug_post_mortem": {
      "default": false,
      "description": "Enable post-mortem debugging: inject `RAY_DEBUG_POST_MORTEM=1` into the Ray cluster configuration. Learn more: [KubeRay docs](https://docs.ray.io/en/latest/ray-observability/ray-distributed-debugger.html)",
      "title": "Enable Debug Post Mortem",
      "type": "boolean"
    },
    "enable_legacy_debugger": {
      "default": false,
      "description": "Enable legacy debugger: inject `RAY_DEBUG=legacy` into the Ray cluster configuration. Learn more: [KubeRay docs](https://docs.ray.io/en/latest/ray-observability/user-guides/debug-apps/ray-debugging.html#using-the-ray-debugger)",
      "title": "Enable Legacy Debugger",
      "type": "boolean"
    }
  },
  "title": "RayResource",
  "type": "object"
}

Fields:

lifecycle (Lifecycle)
timeout (float)
ray_init_options (dict[str, Any])
data_execution_options (RayDataExecutionOptions)
redis_port (int)
dashboard_port (int)
env_vars (dict[str, str] | None)
enable_tracing (bool)
enable_actor_task_logging (bool)
enable_debug_post_mortem (bool)
enable_legacy_debugger (bool)
_context (BaseContext | None)

Attributes¶

lifecycle `pydantic-field` ¶

lifecycle: Lifecycle

Actions to perform during resource setup.

timeout `pydantic-field` ¶

timeout: float = 600.0

Timeout for Ray readiness in seconds

ray_init_options `pydantic-field` ¶

ray_init_options: dict[str, Any]

Additional keyword arguments to pass to ray.init() call, such as runtime_env, num_cpus, etc. Dagster's EnvVar is supported. More details in Ray docs.

data_execution_options `pydantic-field` ¶

data_execution_options: RayDataExecutionOptions

redis_port `pydantic-field` ¶

redis_port: int = 10001

Redis port for connection. Make sure to match with the actual available port.

dashboard_port `pydantic-field` ¶

dashboard_port: int = 8265

Dashboard port for connection. Make sure to match with the actual available port.

env_vars `pydantic-field` ¶

env_vars: dict[str, str] | None

Environment variables to pass to the Ray cluster.

enable_tracing `pydantic-field` ¶

enable_tracing: bool = False

Enable tracing: inject RAY_PROFILING=1 and RAY_task_events_report_interval_ms=0 into the Ray cluster configuration. This allows using ray.timeline() to fetch recorded task events. Learn more: KubeRay docs

enable_actor_task_logging `pydantic-field` ¶

enable_actor_task_logging: bool = False

Enable actor task logging: inject RAY_ENABLE_RECORD_ACTOR_TASK_LOGGING=1 into the Ray cluster configuration.

enable_debug_post_mortem `pydantic-field` ¶

enable_debug_post_mortem: bool = False

Enable post-mortem debugging: inject RAY_DEBUG_POST_MORTEM=1 into the Ray cluster configuration. Learn more: KubeRay docs

enable_legacy_debugger `pydantic-field` ¶

enable_legacy_debugger: bool = False

Enable legacy debugger: inject RAY_DEBUG=legacy into the Ray cluster configuration. Learn more: KubeRay docs

_context `pydantic-field` ¶

_context: BaseContext | None

context `property` ¶

context: BaseContext

host `abstractmethod` `property` ¶

host: str

name `abstractmethod` `property` ¶

name: str

display_name `property` ¶

display_name: str

ray_address `property` ¶

ray_address: str

dashboard_url `property` ¶

dashboard_url: str

runtime_job_id `property` ¶

runtime_job_id: str

Returns the Ray Job ID for the current job which was created with ray.init(). :return:

created `property` ¶

created: bool

ready `property` ¶

ready: bool

connected `property` ¶

connected: bool