write_delta#

Dataset.write_delta(path: str, *, catalog: Catalog | None = None, mode: SaveMode = SaveMode.APPEND, partition_by: List[str] | None = None, storage_options: Dict[str, str] | None = None, schema_mode: str = 'merge', filesystem: pyarrow.fs.FileSystem | None = None, name: str | None = None, description: str | None = None, ray_remote_args: Dict[str, Any] | None = None, concurrency: int | None = None) → None[source]#

Writes the Dataset to a Delta Lake table.

This is a prototype that supports APPEND and OVERWRITE modes. Each write task writes Parquet data files to the table location, and the driver performs a single atomic commit to the Delta transaction log using the deltalake library. If the table doesn’t exist, it’s created.

Tip

For more details on Delta Lake, see https://delta.io/ and the deltalake Python library.

Note

This operation will trigger execution of the lazy transformations performed on this dataset.

Examples

import ray
import pandas as pd
from ray.data import SaveMode

docs = [{"id": i, "title": f"Doc {i}"} for i in range(4)]
ds = ray.data.from_pandas(pd.DataFrame(docs))

# Append (default) - creates the table if it doesn't exist.
ds.write_delta("/tmp/my_delta_table")

# Overwrite the table contents.
ds.write_delta("/tmp/my_delta_table", mode=SaveMode.OVERWRITE)

# Partitioned write.
ds.write_delta("/tmp/my_delta_table", partition_by=["id"])

# A later append with an extra column adds it to the table
# automatically (the default schema_mode="merge") -- existing
# rows read back with None for it.
more_docs = [{"id": 4, "title": "Doc 4", "views": 100}]
ray.data.from_pandas(pd.DataFrame(more_docs)).write_delta(
    "/tmp/my_delta_table"
)

# Reject that same write instead, leaving the table untouched.
ray.data.from_pandas(pd.DataFrame(more_docs)).write_delta(
    "/tmp/my_delta_table", schema_mode="error"
)
Parameters:

  • path – URI of the Delta table (e.g. /tmp/my_table or s3://bucket/my_table). If catalog is set, this is instead the table identifier the catalog resolves (e.g. "main.schema.table").

  • catalog – Optional catalog (e.g. a Unity Catalog connector) that resolves path to its physical location and vends credentials for it. See Catalog. Cannot be combined with filesystem.

  • mode –

    Write mode using the SaveMode enum. Options:

    • SaveMode.APPEND (default): Add new data to the table.

    • SaveMode.OVERWRITE: Replace all existing data in the table.

  • partition_by – Optional list of columns to partition the table by.

  • storage_options – A dictionary of storage options passed to the deltalake library for authentication and configuration (e.g. cloud credentials).

  • schema_mode –

    How an APPEND handles a column present in the data being written but absent from the table’s current schema. Has no effect on SaveMode.OVERWRITE (which always replaces the table’s schema wholesale) or when the table doesn’t exist yet (there’s no existing schema to compare against). One of:

    • "merge" (default): Add the new column to the table before committing the write, like SQL’s ALTER TABLE ... ADD COLUMN. The new column is always added as nullable, and every row written before this reads back with None for it.

    • "error": Reject the write with a ValueError instead, leaving the table’s schema unchanged.

    A column present in both the data and the table, but with an incompatible type (for example, writing a string into a column the table has as an integer), always raises a ValueError – regardless of schema_mode. Only adding a brand-new column is supported; changing an existing column’s type is not.

  • filesystem – Optional PyArrow filesystem used for worker Parquet writes, instead of one built from storage_options or ambient credentials. Cannot be combined with catalog.

  • name – Optional table name recorded in the Delta metadata when a new table is created.

  • description – Optional table description recorded in the Delta metadata when a new table is created.

  • ray_remote_args – kwargs passed to ray.remote() in the write tasks.

  • concurrency – 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. By default, concurrency is dynamically decided based on the available resources.

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

On this page