xpark.dataset.CommonCompute#

class xpark.dataset.CommonCompute(*args, **kwargs)#

Note

Do not construct this class, use the staticmethod instead.

Methods

binary_persist(output_path, *[, filename, ext])

Persist binary data to a storage location and return the URIs.
binary_persist(output_path: str, *, filename: pa.ChunkedArray | None = None, ext: str | None = None) → pa.Array#

Persist binary data to a storage location and return the URIs.

Writes each row of the binarys column to output_path and returns the resulting destination URI for every row. Suitable for offloading large binary blobs (images, audio, video frames, …) from the dataset to a local directory or a remote object store.

Parameters:

  • binarys – A binary column whose values will be persisted to disk.

  • output_path – The directory (or URI prefix) where the blobs will be written. Supports any backend registered with fsspec (e.g. /tmp/data, cos://bucket/prefix, s3://bucket/prefix, hf://repo).

  • filename – Optional per-row filenames. When provided, each value is used as the destination name (a bare name or a relative sub-path). When omitted or null for a row, a random UUID4-based name is generated.

  • ext – Optional file extension applied only to auto-generated UUID names. A leading dot is optional ("bin" and ".bin" are equivalent). Ignored for rows that supply an explicit filename.

Returns:

A string array of destination URIs aligned with the input binarys column. Null inputs produce None entries.

Example

from xpark.dataset import CommonCompute
from xpark.dataset.expressions import col
from xpark.dataset.read_api import from_items
import pyarrow as pa

@udf(return_dtype=pa.binary())
def to_binary(input: pa.Array) -> pa.Array:
    return pa.array([pa.scalar(x.as_py(), type=pa.binary()) for x in input])

ds = from_items([{"data": b"hello", "filename": "hello.txt"}])
ds = ds.with_column("data_binary", to_binary(col("data")))
ds = ds.with_column(
    "data_uri",
    CommonCompute.binary_persist(
        col("data_binary"),
        output_path="/tmp/xpark_test",
        filename=col("filename"),
    )
)
print(ds.take_all())