read_delta#

ray.data.read_delta(path: str | List[str], version: int | None = None, *, storage_options: Dict[str, Any] | None = None, filesystem: pyarrow.fs.FileSystem | None = None, catalog: Catalog | None = None, columns: List[str] | None = None, parallelism: int = -1, num_cpus: float | None = None, num_gpus: float | None = None, memory: float | None = None, shuffle: Literal['files'] | None = None, include_paths: bool = False, concurrency: int | None = None, override_num_blocks: int | None = None, label_selector: Dict[str, str] | None = None, fallback_strategy: List[Dict[str, Any]] | None = None, max_calls: int | None = None, resources: Dict[str, float] | None = None, accelerator_type: str | None = None, runtime_env: Dict[str, Any] | None = None, ray_remote_args: Dict[str, Any] | None = None, **arrow_parquet_args)[source]#

Creates a Dataset from a Delta Lake table.

This reader uses the deltalake library to read the Delta transaction log and constructs a PyArrow dataset that preserves the table’s unified schema, partition information, and column statistics. This enables:

  • Schema evolution support (older files with missing columns are null-filled)

  • Correct handling of cloud storage URIs (Azure, S3, GCS)

  • Column statistics from the Delta log for row-group pruning

  • Authentication via storage_options

Examples

Read a local Delta table:

>>> import ray
>>> ds = ray.data.read_delta("/path/to/delta-table/")

Read from S3 with credentials:

>>> ds = ray.data.read_delta(
...     "s3://bucket/delta-table/",
...     storage_options={
...         "AWS_ACCESS_KEY_ID": "...",
...         "AWS_SECRET_ACCESS_KEY": "...",
...     },
... )

Read from Azure with default credentials:

>>> ds = ray.data.read_delta(
...     "az://container/delta-table/",
...     storage_options={"use_azure_cli": "true"},
... )
Parameters:

  • path (str | List[str]) – A single path to a Delta Lake table. Multiple tables are not supported.

  • version (int | None) – The version of the Delta Lake table to read. If not specified, the latest version is read.

  • storage_options (Dict[str, Any] | None) – A dictionary of storage options passed to the deltalake library for authentication and configuration. Supported keys depend on the storage backend: S3 options, Azure options, GCS options.

  • filesystem (pyarrow.fs.FileSystem | None) – The PyArrow filesystem implementation to read from. These filesystems are specified in the pyarrow docs. Specify this parameter if you need to provide specific configurations to the filesystem. By default, the filesystem is automatically selected based on the scheme of the paths. For example, if the path begins with s3://, the S3FileSystem is used. If None, this function uses a system-chosen implementation.

  • catalog (Catalog | None) – An optional Catalog (e.g. DatabricksUnityCatalog) used to authenticate access. When provided, path is interpreted as a catalog table identifier (e.g. "catalog.schema.table") rather than a filesystem path, and the catalog resolves the physical location and credentials.

  • columns (List[str] | None) – A list of column names to read. Only the specified columns are read during the file scan.

  • parallelism (int) – This argument is deprecated. Use override_num_blocks argument.

  • num_cpus (float | None) – The number of CPUs to reserve for each parallel read worker.

  • num_gpus (float | None) – The number of GPUs to reserve for each parallel read worker. For example, specify num_gpus=1 to request 1 GPU for each parallel read worker.

  • memory (float | None) – The heap memory in bytes to reserve for each parallel read worker.

  • shuffle (Literal['files'] | None) – If setting to “files”, randomly shuffle input files order before read. Defaults to not shuffle with None.

  • include_paths (bool) – If True, include the path to each file. File paths are stored in the 'path' column.

  • concurrency (int | None) – The maximum number of Ray tasks to run concurrently. Set this to control number of tasks to run concurrently. This doesn’t change the total number of tasks run or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources.

  • override_num_blocks (int | None) – Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn’t manually set this value in most cases.

  • label_selector (Dict[str, str] | None) – Labels required on the node where each read task runs.

  • fallback_strategy (List[Dict[str, Any]] | None) – Alternative label requirements that Ray tries in order if label_selector can’t be satisfied.

  • max_calls (int | None) – The maximum number of read tasks a worker runs before exiting.

  • resources (Dict[str, float] | None) – Custom resources to reserve for each read task, expressed as a mapping from resource name to quantity.

  • accelerator_type (str | None) – The accelerator type required for each read task.

  • runtime_env (Dict[str, Any] | None) – The runtime environment to use for each read task.

  • ray_remote_args (Dict[str, Any] | None) – Additional options passed to ray.remote() for each read task. This argument is deprecated and will be removed in Ray 2.64. Use the named remote parameters instead.

  • **arrow_parquet_args – Other parquet read options to pass to PyArrow. For the full set of arguments, see the PyArrow API

Returns:

Dataset producing records read from the specified Delta Lake table.

PublicAPI (alpha): This API is in alpha and may change before becoming stable.

On this page