Creating Datasets¶

A Dataset can be created from:

generated synthetic data,
local and distributed in-memory data, and
local and external storage systems (local disk, cloud storage, HDFS, etc.).

Creation from existing in-memory data is enabled via Datasets’ integrations with familiar single-node data libraries (Pandas, NumPy, Arrow) and distributed data processing frameworks (Dask, Spark, Modin, Mars). Creating datasets from persistent storage is enabled by Datasets’ support for reading many common file formats (Parquet, CSV, JSON, NPY, text, binary).

A Dataset can hold plain Python objects (simple datasets), Arrow records (Arrow datasets), or Pandas records (Pandas datasets). These records are grouped into one or more data blocks, and these blocks can be spread across multiple Ray nodes. Simple datasets are represented by simple blocks (lists of Python objects), Arrow datasets are represented by Arrow blocks ( Arrow Tables ), and Pandas datasets are represented by Pandas blocks ( Pandas DataFrames ).

The method of creating the dataset will determine the format of its internal block representation.

This guide surveys the many ways to create a Dataset. If none of these meet your needs, please reach out on Discourse or open a feature request on the Ray GitHub repo, and check out our guide for implementing a custom Datasets datasource if you’re interested in rolling your own integration!

Generating Synthetic Data¶

Using Datasets with small, generated data is a great way to test out some of Datasets’ features. Each of these synthetic data generators will generate data in Ray tasks that will execute in parallel and will be load-balanced across your Ray cluster; the Dataset will hold a set of futures representing the return values of those tasks, serving as pointers to a collection of distributed data blocks.

Int Range

Create a Dataset from a range of integers.

# Create a Dataset of Python objects.
ds = ray.data.range(10000)
# -> Dataset(num_blocks=200, num_rows=10000, schema=<class 'int'>)

ds.take(5)
# -> [0, 1, 2, 3, 4]

Tabular Range

Create an Arrow (tabular) Dataset from a range of integers, with a single column containing this integer range.

# Create a Dataset of Arrow records.
ds = ray.data.range_table(10000)
# -> Dataset(num_blocks=200, num_rows=10000, schema={value: int64})

ds.take(5)
# -> [{'value': 0}, {'value': 1}, {'value': 2}, {'value': 3}, {'value': 4}]

Tensor Range

Create a tensor dataset from a range of integers, packing this integer range into tensors of the provided shape.

# Create a Dataset of tensors.
ds = ray.data.range_tensor(100 * 64 * 64, shape=(64, 64))
# -> Dataset(
#       num_blocks=200,
#       num_rows=409600,
#       schema={value: <ArrowTensorType: shape=(64, 64), dtype=int64>}
#    )

ds.take(2)
# -> [array([[0, 0, 0, ..., 0, 0, 0],
#         [0, 0, 0, ..., 0, 0, 0],
#         [0, 0, 0, ..., 0, 0, 0],
#         ...,
#         [0, 0, 0, ..., 0, 0, 0],
#         [0, 0, 0, ..., 0, 0, 0],
#         [0, 0, 0, ..., 0, 0, 0]]),
#  array([[1, 1, 1, ..., 1, 1, 1],
#         [1, 1, 1, ..., 1, 1, 1],
#         [1, 1, 1, ..., 1, 1, 1],
#         ...,
#         [1, 1, 1, ..., 1, 1, 1],
#         [1, 1, 1, ..., 1, 1, 1],
#         [1, 1, 1, ..., 1, 1, 1]])]

From Local and Distributed In-Memory Data¶

Datasets can be constructed from existing in-memory data. In addition to being able to construct a Dataset from plain Python objects, Datasets also interoperates with popular single-node libraries (Pandas, NumPy, Arrow) as well as distributed frameworks (Dask, Spark, Modin, Mars).

Integration with Single-Node Data Libraries¶

In this section, we demonstrate creating a Dataset from single-node in-memory data.

Pandas

Create a Dataset from a Pandas DataFrame. This constructs a Dataset backed by a single Pandas DataFrame block.

import pandas as pd

# Create a tabular Dataset from a Pandas DataFrame.
df = pd.DataFrame({"col1": list(range(10000)), "col2": list(map(str, range(10000)))})
ds = ray.data.from_pandas(df)
# -> Dataset(num_blocks=1, num_rows=10000, schema={col1: int64, col2: object})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

We can also build a Dataset from more than one Pandas DataFrame, where each said DataFrame will become a block in the Dataset.

import pandas as pd

data = list(range(10000))
num_chunks = 10
chunk_size = len(data) // num_chunks
chunks = [data[i : i + chunk_size] for i in range(0, len(data), chunk_size)]
dfs = [
    pd.DataFrame({"col1": list(chunk), "col2": list(map(str, chunk))})
    for chunk in chunks
]
# Create a tabular Dataset from multiple Pandas DataFrames.
ds = ray.data.from_pandas(dfs)
# -> Dataset(num_blocks=10, num_rows=10000, schema={col1: int64, col2: object})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

NumPy

Create a Dataset from a NumPy ndarray. This constructs a Dataset backed by a single-column Arrow table block; the outer dimension of the ndarray will be treated as the row dimension, and the column with have name "value".

import numpy as np

# Create a tensor Dataset from a 1D NumPy ndarray.
arr = np.arange(100)
ds = ray.data.from_numpy(arr)
# -> Dataset(
#        num_blocks=1,
#        num_rows=100,
#        schema={value: <ArrowTensorType: shape=(), dtype=int64>},
#    )

# Each element is a scalar ndarray.
ds.show(3)
# -> {'value': array(0)}
# -> {'value': array(1)}
# -> {'value': array(2)}

# Create a tensor Dataset from a 3D NumPy ndarray.
arr = np.ones((3, 4, 4))
# The outer dimension is treated as the row dimension.
ds = ray.data.from_numpy(arr)
# -> Dataset(
#        num_blocks=1,
#        num_rows=3,
#        schema={value: <ArrowTensorType: shape=(4, 4), dtype=int64>},
#    )

ds.show(2)
# -> {'value': array([[1., 1., 1., 1.],
#        [1., 1., 1., 1.],
#        [1., 1., 1., 1.],
#        [1., 1., 1., 1.]])}
# -> {'value': array([[1., 1., 1., 1.],
#        [1., 1., 1., 1.],
#        [1., 1., 1., 1.],
#        [1., 1., 1., 1.]])}

We can also build a Dataset from more than one NumPy ndarray, where each said ndarray will become a single-column Arrow table block in the Dataset.

import numpy as np

# Create a tensor Dataset from multiple 3D NumPy ndarray.
arrs = [np.random.rand(2, 4, 4) for _ in range(4)]
# The outer dimension is treated as the row dimension.
ds = ray.data.from_numpy(arrs)
# -> Dataset(
#        num_blocks=4,
#        num_rows=8,
#        schema={value: <ArrowTensorType: shape=(4, 4), dtype=int64>},
#    )

ds.show(2)
# -> {'value': array([[0.06587483, 0.67808656, 0.76461924, 0.83428549],
#        [0.04932103, 0.25112165, 0.26476714, 0.24599738],
#        [0.67624391, 0.58689537, 0.12594709, 0.94663371],
#        [0.32435665, 0.97719096, 0.03234169, 0.71563231]])}
# -> {'value': array([[0.98570318, 0.65956399, 0.82168898, 0.09798336],
#        [0.22426704, 0.34209978, 0.02605247, 0.48200137],
#        [0.17312096, 0.38789983, 0.42663678, 0.92652456],
#        [0.80787394, 0.92437162, 0.11185822, 0.3319638 ]])}

Arrow

Create a Dataset from an Arrow Table. This constructs a Dataset backed by a single Arrow Table block.

import pyarrow as pa

# Create a tabular Dataset from an Arrow Table.
t = pa.table({"col1": list(range(10000)), "col2": list(map(str, range(10000)))})
ds = ray.data.from_arrow(t)
# -> Dataset(num_blocks=1, num_rows=10000, schema={col1: int64, col2: string})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

We can also build a Dataset from more than one Arrow Table, where each said Table will become a block in the Dataset.

import arrow as pd

data = list(range(10000))
num_chunks = 10
chunk_size = len(data) // num_chunks
chunks = [data[i : i + chunk_size] for i in range(0, len(data), chunk_size)]
ts = [
    pa.table({"col1": list(chunk), "col2": list(map(str, chunk))})
    for chunk in chunks
]
# Create a tabular Dataset from multiple Arrow Tables.
ds = ray.data.from_arrow(ts)
# -> Dataset(num_blocks=10, num_rows=10000, schema={col1: int64, col2: string})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

Python Objects

Create a Dataset from a list of Python objects; since each object in this particular list is a dictionary, Datasets will treat this list as a list of tabular records, and will construct an Arrow Dataset.

# Create a Dataset of tabular (Arrow) records.
ds = ray.data.from_items([{"col1": i, "col2": str(i)} for i in range(10000)])
# -> Dataset(num_blocks=200, num_rows=10000, schema={col1: int64, col2: string})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

Integration with Distributed Data Processing Frameworks¶

In addition to working with single-node in-memory data, Datasets can be constructed from distributed (multi-node) in-memory data, interoperating with popular distributed data processing frameworks such as Dask, Spark, Modin, and Mars.

The common paradigm used by these conversions is to send out Ray tasks converting each Dask/Spark/Modin/Mars data partition to a format that Datasets can understand (if needed), and using the futures representing the return value of those conversion tasks as the Dataset block futures. If the upstream framework’s data partitions are already in a format that Datasets understands (e.g. Arrow or Pandas), Datasets will elide the conversion task and will instead reinterpret those data partitions directly as its blocks.

Note

These data processing frameworks must be running on Ray in order for these Datasets integrations to work. See how these frameworks can be run on Ray in our data processing integrations docs.

Dask

Create a Dataset from a Dask DataFrame. This constructs a Dataset backed by the distributed Pandas DataFrame partitions that underly the Dask DataFrame.

Note

This conversion should have near-zero overhead: it involves zero data copying and zero data movement. Datasets simply reinterprets the existing Dask DataFrame partitions as Ray Datasets partitions without touching the underlying data.

import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({"col1": list(range(10000)), "col2": list(map(str, range(10000)))})
ddf = dd.from_pandas(df, npartitions=4)
# Create a tabular Dataset from a Dask DataFrame.
ds = ray.data.from_dask(ddf)
# -> Dataset(num_blocks=10, num_rows=10000, schema={col1: int64, col2: object})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

Spark

Create a Dataset from a Spark DataFrame. This constructs a Dataset backed by the distributed Spark DataFrame partitions that underly the Spark DataFrame. When this conversion happens, Spark-on-Ray (RayDP) will save the Spark DataFrame partitions to Ray’s object store in the Arrow format, which Datasets will then interpret as its blocks.

import raydp

spark = raydp.init_spark(app_name="Spark -> Datasets Example",
                         num_executors=2,
                         executor_cores=2,
                         executor_memory="500MB")
df = spark.createDataFrame([(i, str(i)) for i in range(10000)], ["col1", "col2"])
# Create a tabular Dataset from a Spark DataFrame.
ds = ray.data.from_dask(df)
# -> Dataset(num_blocks=10, num_rows=10000, schema={col1: int64, col2: string})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

Modin

Create a Dataset from a Modin DataFrame. This constructs a Dataset backed by the distributed Pandas DataFrame partitions that underly the Modin DataFrame.

Note

This conversion should have near-zero overhead: it involves zero data copying and zero data movement. Datasets simply reinterprets the existing Modin DataFrame partitions as Ray Datasets partitions without touching the underlying data.

import modin.pandas as md

df = pd.DataFrame({"col1": list(range(10000)), "col2": list(map(str, range(10000)))})
mdf = md.DataFrame(df)
# Create a tabular Dataset from a Modin DataFrame.
ds = ray.data.from_modin(mdf)
# -> Dataset(num_blocks=8, num_rows=10000, schema={col1: int64, col2: object})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

Mars

Create a Dataset from a Mars DataFrame. This constructs a Dataset backed by the distributed Pandas DataFrame partitions that underly the Mars DataFrame.

Note

This conversion should have near-zero overhead: it involves zero data copying and zero data movement. Datasets simply reinterprets the existing Mars DataFrame partitions as Ray Datasets partitions without touching the underlying data.

import mars
import mars.dataframe as md

cluster = mars.new_cluster_in_ray(worker_num=2, worker_cpu=1)

df = pd.DataFrame({"col1": list(range(10000)), "col2": list(map(str, range(10000)))})
mdf = md.DataFrame(df, num_partitions=8)
# Create a tabular Dataset from a Mars DataFrame.
ds = ray.data.from_mars(mdf)
# -> Dataset(num_blocks=8, num_rows=10000, schema={col1: int64, col2: object})

ds.show(3)
# -> {'col1': 0, 'col2': '0'}
# -> {'col1': 1, 'col2': '1'}
# -> {'col1': 2, 'col2': '2'}

From Torch and TensorFlow¶

PyTorch

If you already have a Torch dataset available, you can create a Ray Dataset using SimpleTorchDatasource.

Warning

SimpleTorchDatasource doesn’t support parallel reads. You should only use this datasource for small datasets like MNIST or CIFAR.

import ray.data
from ray.data.datasource import SimpleTorchDatasource
import torchvision

dataset_factory = lambda: torchvision.datasets.MNIST("data", download=True)
dataset = ray.data.read_datasource(
    SimpleTorchDatasource(), parallelism=1, dataset_factory=dataset_factory
)
dataset.take(1)
# (<PIL.Image.Image image mode=L size=28x28 at 0x1142CCA60>, 5)

TensorFlow

If you already have a TensorFlow dataset available, you can create a Ray Dataset using SimpleTensorFlowDatasource.

Warning

SimpleTensorFlowDatasource doesn’t support parallel reads. You should only use this datasource for small datasets like MNIST or CIFAR.

import ray.data
from ray.data.datasource import SimpleTensorFlowDatasource
import tensorflow_datasets as tfds

def dataset_factory():
    return tfds.load("cifar10", split=["train"], as_supervised=True)[0]

dataset = ray.data.read_datasource(
    SimpleTensorFlowDatasource(),
    parallelism=1,
    dataset_factory=dataset_factory
)
features, label = dataset.take(1)[0]
features.shape  # TensorShape([32, 32, 3])
label  # <tf.Tensor: shape=(), dtype=int64, numpy=7>

From 🤗 (Hugging Face) Datasets¶

You can convert 🤗 Datasets into Ray Datasets by using from_huggingface. This function accesses the underlying Arrow table and converts it into a Ray Dataset directly.

Warning

from_huggingface doesn’t support parallel reads. This will not usually be an issue with in-memory 🤗 Datasets, but may fail with large memory-mapped 🤗 Datasets. 🤗 IterableDataset objects are not supported.

import ray.data
from datasets import load_dataset

hf_datasets = load_dataset("wikitext", "wikitext-2-raw-v1")
ray_datasets = ray.data.from_huggingface(hf_datasets)
ray_datasets["train"].take(2)
# [{'text': ''}, {'text': ' = Valkyria Chronicles III = \n'}]

Ray 3.0.0.dev0

Creating Datasets¶

Generating Synthetic Data¶

From Local and Distributed In-Memory Data¶

Integration with Single-Node Data Libraries¶

Integration with Distributed Data Processing Frameworks¶

Reading Files From Storage¶

Parallel + Distributed Reading¶

Deferred Read Task Execution¶

Supported File Formats¶

Reading from Remote Storage¶

From Torch and TensorFlow¶

From 🤗 (Hugging Face) Datasets¶