&`it,ddlZddlZddlZddlZddlZddlZddlZddlmZm Z m Z m Z m Z m Z mZmZmZmZmZmZmZmZddlZddlZddlmZddlmZddlmZddl m!Z!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(m)Z)m*Z*dd l+m,Z,dd l-m.Z.dd l/m0Z0dd l1m2Z2dd l3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBddlCmDZDddlEmFZFddlGmHZHddlImJZJddlKmLZLddlMmNZNddlOmPZPmQZQmRZRmSZSddlTmUZUddlVmWZWddlXmYZYddlZm[Z[m\Z\m]Z]m^Z^m_Z_m`Z`dd lamZbmcZcdd!ldmeZedd"lfmgZgdd#lhmiZidd$ljmkZkmlZldd%lmmnZndd&lompZpdd'lqmrZrdd(lsmtZtmuZudd)lvmwZwmxZxmyZydd*lzm{Z{m|Z|m}Z}m~Z~mZdd+lmZmZmZmZmZmZmZmZdd,lmZmZmZmZmZmZmZmZmZdd-lmZdd.lmZmZmZmZdd/lmZmZdd0lmZdd1lmZdd2lmZdd3lmZdd4lmZdd5lmZmZmZdd6lmZdd7lmZdd8lmZerBddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZdd9lmZdd:lCmZmZdd;lmZddZed?ed?e ed?ffZed@e ed@ffZedAZee edBfefZeedCefZ dDZdEZdFZdGZdHZdIZdJZdKZdLZeGdMdNZeGdOdPee eZedQRGdSdTZdUedVdWfdXZdUedYeefdZZdUefd[ZdS)\N) TYPE_CHECKINGAnyCallableDictGenericIterableIteratorListLiteralMappingOptionalTupleTypeVarUnion) usage_lib)tabulate)ArrowTensorTypeV2,get_arrow_extension_fixed_shape_tensor_types)ComputeStrategy)BigQueryDatasink)ClickHouseDatasinkClickHouseTableSettingsSinkMode) CSVDatasink)IcebergDatasink) ImageDatasink) JSONDatasink) LanceDatasink) MongoDatasink) NumpyDatasink)ParquetDatasink) SQLDatasink)TFRecordDatasink)WebDatasetDatasink) _equalize) RefBundle)(_ref_bundles_iterator_to_block_refs_list) memory_stringDataIteratorImpl)StreamSplitDataIterator) LogicalPlan)RandomizeBlocks RandomShuffle RepartitionSort)Count) InputData)Join)FilterFlatMap MapBatchesMapRowsProjectStreamingRepartition)rZip)Limit)StreamingSplit)Write)PandasBlockBuilderPandasBlockSchema) ExecutionPlan)SortKey)cached_remote_fn) _get_num_rows_split_at_indices) DatasetStatsDatasetStatsSummary _StatsManager) AllToAllAPIConsumptionAPI_validate_rows_per_file_argsget_compute_strategy"merge_resources_to_ray_remote_args) AggregateFn AggregateFnV2MaxMeanMinStdSumUnique) VALID_BATCH_FORMATSBlock BlockAccessor DataBatchDataBatchColumnTUUserDefinedFunction_apply_batch_format) DataContext) ConnectionDatasinkFilenameProviderSaveMode) WriteResult_gen_datasink_write_result) _FileDatasink)DataType) DataIterator)RandomAccessDataset) ObjectRef) Deprecated DeveloperAPI PublicAPI)NodeAffinitySchedulingStrategy)Template)repr_with_fallback) schema_pb2)Executor NodeIdStr GroupedData)DatasetSummary)ExprStarExprcol!__ray_train_test_split_is_train__ tf.TypeSpecz tf.Tensor CollatedDataz torch.Tensorz torch.devicezBasic Transformationsz%Sorting, Shuffling and Repartitioningz$Splitting, Merging, Joining DatasetszGrouped and Global AggregationszConsuming DatazI/O and ConversionzInspecting Metadata Execution Expressionsc%x4eZdZdZdedefdZe dVddded e e d dfd Z e e ddddddddddd deeeefgeeeffde ede eede eeefde eede eeefde ede ede ede eeeeefeeeeffde egeeeffd dfdZeddde efdZde efdZeed dd e efd!Zed e efd"Zd efd#Ze e ddd$dddddddddddd%de e!e!fd&eede"d$fde ed'e ed(ede eede eeefde eede eeefde ede ede ede eeeeefeeeeffd)ede egeeeffd df d*Z#de e!e!fd&eede"d$fde ed'e ed(ede eede eeefde eede eeefde ede ede ede eeeeefeeeeffd)ede egeeefffd+Z$e e%d,-d.ed/e&d dfd0Z'ed12e e d3ddd4d5edee!ge(fd'e ede ede ed df d6Z)e e ddd7d8e*ede ede ed dfd9Z+e e ddd7d8eee*efdeeefde ed dfd:Z,e e dd;dZ.e e  dWddddddddddd de e eeefefd/e eee&fdeeefde eede eeefde eede eeefde ede ede ede eeeeefeeeeffde egeeeffd dfd?Z/e e0  dWdddd@dAe edBe edCedDe e*edEed df dFZ1e2e e0 dddGdHe edAe ed dfdIZ3e2e e0 ddJdHe ed dfdKZ4e e ddJdLedHe ed dfdMZ5e6e e7 dddNdOedPedQe e*dRd e*e8fdSZ9e6e e7 dddNdOedPedQe e*ed e*dTfdUZ:e6e e7 dVe*ed e*dTfdWZ;e6e e7 dXe*ed e*dTfdYZd[ed dfd`Z?d[eddd efdaZ@e d,e7bdcddddd[edee"dfdge edHe ed edhf diZAe e7 dje*dd dfdkZBe2e e7  dXddddmdddnedoedpeedqe eedre edse edte edue eeefdved dfdwZCe2e eD  dYdxeee*edfdoe ed dyfdzZEe2e6e eD dZd{ed|ed e*efd}ZFe2e6e eD d~eGd eeeeefffdZHe2e6e eD  d[dpe eee*efd|ed eeeeefffdZIe2e6e eD  d[dpe eee*efd|ed eeeeefffdZJe2e6e eD  d[dpe eee*efd|ed eeeeefffdZKe2e6e eD  d[dpe eee*efd|ed eeeeefffdZLe2e6e eD  d\dpe eee*efded|ed eeeeefffdZMe2e6e eDd,- dWde e*ede eeNeege*eOffd dfdZPe2e e0  dVdxeee*efdeee*efde*eeefd dfdZQe e7 dje*dd dfdZRe e ded dfdZSe6e eT  d]d$dd&ed'e ed e!fdZUe6e eT d]ded e*eeeffdZVe6e eT dYde ed e*eeeffdZWe6e eT d]ded dfdZXe6ddde eY d efdZZe6dddde eY d^ded e dfdZ[e6dddde eY d^ded e e*efdZ\e eY d efdZ]e6e eY d efdZ^e6e eY d e*efdZ_e6e e` dddddddddddeajbd dede e*ede ddede eeefde ecde egeeeffde ede edeeefde ede edead dfdZde6e e` dddddddddeajbd dede ddede eeefde ecde egeeeffde edeeefde ede edead dfdZee6e d,e`bddeajbdddddfdede eeefde eeefddde dde eeefde eeefdeeefde ed dfdZfe d,e`be6 d_ddddddeajbdœded{edede ddede eeefde ecdeeefde edead dfdĄZge6e e` dddddddddeajbdŜ dede ddede eeefde ecde egeeeffde edeeefde ede edead dfdDŽZhe6e e` dddddddddeajbdȜ dede dde ddede eeefde ecde edeeefde ede edead dfd˄Zie6e d,e`bdddddddddeajbd̜ dede ddede eeefde ecde edeeefde eeeejekfde ede edead dfd΄Zle6e e` ddddddddeajbdϜ ded{ede ddede eeefde ecde edeeefde ede edead dfdЄZme6 dWdedegenfde eeefde ed df dӄZoe6dddԜdededeeefde efdׄZpe d,e`be6 dWdedededeeefde ed df dۄZqe6 d`dededede edeeefde ed dfdZre6esjtdddddddddededesde dde eeefde eeefde eude edeeefde ed dfdZve6ddddddddddede dde"ddedede ede eeefdeeefde ed dfdZwe6ddddԜdexdeeefde ed dfdZye6dde eT d e8fdZze6e eT d eeeeffdZ{e6e eT ddd$dddddded&e ed'e edede ede ede ee!ge|fd ee!fdZ}e6e eT dddddddddded&e ede edeedffdee~e"dfde eeeejfge|fdede ede ed eefdZe6eddddddd ded&e ede ed eed ffdede ede ed eefd Ze6e e` dddddddddd d eee*efdeee*efdeee*efded&edede ede ededeedffdedeedffdedeedffd dfdZe6de e` dadZe6de e`  d[deddeeefeeeedfded dfdZe6de e` dbdZe6de e` dcd!Ze6de e` ddd%Ze6de e` dYded dfd&Ze6ded e*edfd'Zedd(d{e ed e*eejfd)Ze6ded e*ed*fd+Ze6d, dYdxed-e ed efd.Ze6d/d0e e ded1Ze eY d efd2Ze eYd,-d3Zd efd4Ze6ded eefd5Zee6dd e*eefd6Zed efd7Zed efd8Zeed9ed dfd:Zeed efd;Zdd`. Datasets can be created in multiple ways: * from external storage systems such as local disk, S3, HDFS etc. via the ``read_*()`` APIs. * from existing memory data via ``from_*()`` APIs * from synthetic data via ``range_*()`` APIs The (potentially processed) Dataset can be saved back to external storage systems via the ``write_*()`` APIs. Examples: .. testcode:: :skipif: True import ray # Create dataset from synthetic data. ds = ray.data.range(1000) # Create dataset from in-memory data. ds = ray.data.from_items( [{"col1": i, "col2": i * 2} for i in range(1000)] ) # Create dataset from external storage system. ds = ray.data.read_parquet("s3://bucket/path") # Save dataset back to external storage system. ds.write_csv("s3://bucket/output") Dataset has two kinds of operations: transformation, which takes in Dataset and outputs a new Dataset (e.g. :py:meth:`.map_batches()`); and consumption, which produces values (not a data stream) as output (e.g. :meth:`.iter_batches()`). Dataset transformations are lazy, with execution of the transformations being triggered by downstream consumption. Dataset supports parallel processing at scale: * transformations such as :py:meth:`.map_batches()` * aggregations such as :py:meth:`.min()`/:py:meth:`.max()`/:py:meth:`.mean()`, * grouping via :py:meth:`.groupby()`, * shuffling operations such as :py:meth:`.sort()`, :py:meth:`.random_shuffle()`, and :py:meth:`.repartition()` * joining via :py:meth:`.join()` Examples: >>> import ray >>> ds = ray.data.range(1000) >>> # Transform batches (Dict[str, np.ndarray]) with map_batches(). >>> ds.map_batches(lambda batch: {"id": batch["id"] * 2}) # doctest: +ELLIPSIS MapBatches() +- Dataset(num_rows=1000, schema={id: int64}) >>> # Compute the maximum. >>> ds.max("id") 999 >>> # Shuffle this dataset randomly. >>> ds.random_shuffle() # doctest: +ELLIPSIS RandomShuffle +- Dataset(num_rows=1000, schema={id: int64}) >>> # Sort it back in order. >>> ds.sort("id") # doctest: +ELLIPSIS Sort +- Dataset(num_rows=1000, schema={id: int64}) Both unexecuted and materialized Datasets can be passed between Ray tasks and actors without incurring a copy. Dataset supports conversion to/from several more featureful dataframe libraries (e.g., Spark, Dask, Modin, MARS), and are also compatible with distributed TensorFlow / PyTorch. plan logical_planc>t|tsJt|tjd||_||_|j|d|_d|_ | tj dS)zConstruct a Dataset (internal API). The constructor is not part of the Dataset API. Use the ``ray.data.*`` read methods to construct a dataset. datasetN) isinstancer@typerrecord_library_usage_plan _logical_planlink_logical_plan_current_executor _write_ds _set_uuidrGgen_dataset_id_from_stats_actor)selfrrs d/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/dataset.py__init__zDataset.__init__s$ ..::T ::.&y111 ) $$\2228< }DFFGGGGGFNds _deep_copy_asreturnc|st|}|r(||j|jS||j|jSN)rr deep_copyrcopy)rrrs rrz Dataset.copysa r((C  :3rx))++R-=>> >3rx}}(899 9r) api_group) computefn_args fn_kwargsfn_constructor_argsfn_constructor_kwargsnum_cpusnum_gpusmemory concurrencyray_remote_args_fnfnrrrrrrrrrrc t|||| }t||| | } |j} t |jj||||||| |  }t||j}t| |S)avApply the given function to each row of this dataset. Use this method to transform your data. To learn more, see :ref:`Transforming rows `. You can use either a function or a callable class to perform the transformation. For functions, Ray Data uses stateless Ray tasks. For classes, Ray Data uses stateful Ray actors. For more information, see :ref:`Stateful Transforms `. .. tip:: If your transformation is vectorized like most NumPy or pandas operations, :meth:`~Dataset.map_batches` might be faster. .. warning:: Specifying both ``num_cpus`` and ``num_gpus`` for map tasks is experimental, and may result in scheduling or stability issues. Please `report any issues `_ to the Ray team. Examples: .. testcode:: import os from typing import Any, Dict import ray def parse_filename(row: Dict[str, Any]) -> Dict[str, Any]: row["filename"] = os.path.basename(row["path"]) return row ds = ( ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple", include_paths=True) .map(parse_filename) ) print(ds.schema()) .. testoutput:: Column Type ------ ---- image ArrowTensorTypeV2(shape=(32, 32, 3), dtype=uint8) path string filename string Time complexity: O(dataset size / parallelism) Args: fn: The function to apply to each row, or a class type that can be instantiated to create such a callable. compute: The compute strategy to use for the map operation. * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks. * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks. * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers. * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n, initial_size=initial)`` to use an autoscaling actor pool from ``m`` to ``n`` workers, with an initial size of ``initial``. fn_args: Positional arguments to pass to ``fn`` after the first argument. These arguments are top-level arguments to the underlying Ray task. fn_kwargs: Keyword arguments to pass to ``fn``. These arguments are top-level arguments to the underlying Ray task. fn_constructor_args: Positional arguments to pass to ``fn``'s constructor. You can only provide this if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. fn_constructor_kwargs: Keyword arguments to pass to ``fn``'s constructor. This can only be provided if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. num_cpus: The number of CPUs to reserve for each parallel map worker. num_gpus: The number of GPUs to reserve for each parallel map worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel map worker. memory: The heap memory in bytes to reserve for each parallel map worker. concurrency: This argument is deprecated. Use ``compute`` argument. ray_remote_args_fn: A function that returns a dictionary of remote args passed to each map worker. The purpose of this argument is to generate dynamic arguments for each actor/task, and will be called each time prior to initializing the worker. Args returned from this dict will always override the args in ``ray_remote_args``. Note: this is an advanced, experimental feature. ray_remote_args: Additional resource requirements to request from Ray for each map worker. See :func:`ray.remote` for details. .. seealso:: :meth:`~Dataset.flat_map` Call this method to create new rows from existing ones. Unlike :meth:`~Dataset.map`, a function passed to :meth:`~Dataset.flat_map` can return multiple rows. :meth:`~Dataset.map_batches` Call this method to transform batches of data. rrr)rrrrrrray_remote_args) rKrLrrr7rdagr,contextr)rrrrrrrrrrrrrrmap_oprs rmapz Dataset.map$sl'  3#    =       z     "  3"71+    #64<88 t\***rzUse set_name() insteadT)messagewarningnamec0||dSr)set_namerrs r _set_namezDataset._set_names drc||j_dS)zQSet the name of the dataset. Used as a prefix for metrics tags. Nr _dataset_namers rrzDataset.set_names $(    rzUse name() insteadc|jSr)rrs r_namez Dataset._names yrc|jjS)z%Returns the user-defined dataset namerrs rrz Dataset.namesz''rc4|jS)ziUnique ID of the dataset, including the dataset name, UUID, and current execution index. )rget_dataset_idrs rrzDataset.get_dataset_idsz((***rdefault batch_sizer batch_formatzero_copy_batchrrrrrrrrudf_modifying_row_countrrrrrc | duo| dk}|r||dkrtdt|tr|dkrtd|j|f|||||||| | | | | ||d|S)a=*Apply the given function to batches of data. This method is useful for preprocessing data and performing inference. To learn more, see :ref:`Transforming batches `. You can use either a function or a callable class to perform the transformation. For functions, Ray Data uses stateless Ray tasks. For classes, Ray Data uses stateful Ray actors. For more information, see :ref:`Stateful Transforms `. .. tip:: To understand the format of the input to ``fn``, call :meth:`~Dataset.take_batch` on the dataset to get a batch in the same format as will be passed to ``fn``. .. note:: ``fn`` should generally avoid modifying data buffers behind its input since these could be zero-copy views into the underlying object residing inside Ray's Object Store. To perform any modifications it's recommended to copy the data you want to modify. In rare cases when you can't copy inside your UDF, you can instead specify ``zero_copy_batch=False`` and then Ray Data will copy the *whole* batch for you, providing ``fn`` with a copy rather than a zero-copy view. .. warning:: Specifying both ``num_cpus`` and ``num_gpus`` for map tasks is experimental, and may result in scheduling or stability issues. Please `report any issues `_ to the Ray team. Examples: Call :meth:`~Dataset.map_batches` to transform your data. .. testcode:: from typing import Dict import numpy as np import ray def add_dog_years(batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]: batch["age_in_dog_years"] = 7 * batch["age"] return batch ds = ( ray.data.from_items([ {"name": "Luna", "age": 4}, {"name": "Rory", "age": 14}, {"name": "Scout", "age": 9}, ]) .map_batches(add_dog_years) ) ds.show() .. testoutput:: {'name': 'Luna', 'age': 4, 'age_in_dog_years': 28} {'name': 'Rory', 'age': 14, 'age_in_dog_years': 98} {'name': 'Scout', 'age': 9, 'age_in_dog_years': 63} If your function returns large objects, yield outputs in chunks. .. testcode:: from typing import Dict import ray import numpy as np def map_fn_with_large_output(batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]: for i in range(3): yield {"large_output": np.ones((100, 1000))} ds = ( ray.data.from_items([1]) .map_batches(map_fn_with_large_output) ) If you require stateful transformation, use Python callable class. Here is an example showing how to use stateful transforms to create model inference workers, without having to reload the model on each call. .. testcode:: from typing import Dict import numpy as np import torch import ray class TorchPredictor: def __init__(self): self.model = torch.nn.Identity().cuda() self.model.eval() def __call__(self, batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]: inputs = torch.as_tensor(batch["data"], dtype=torch.float32).cuda() with torch.inference_mode(): batch["output"] = self.model(inputs).detach().cpu().numpy() return batch ds = ( ray.data.from_numpy(np.ones((32, 100))) .map_batches( TorchPredictor, # Two workers with one GPU each compute=ray.data.ActorPoolStrategy(size=2), # Batch size is required if you're using GPUs. batch_size=4, num_gpus=1 ) ) To learn more, see :ref:`End-to-end: Offline Batch Inference `. Args: fn: The function or generator to apply to a record batch, or a class type that can be instantiated to create such a callable. Note ``fn`` must be pickle-able. batch_size: The desired number of rows in each batch, or ``None`` to use entire blocks as batches (blocks may contain different numbers of rows). The actual size of the batch provided to ``fn`` may be smaller than ``batch_size`` if ``batch_size`` doesn't evenly divide the block(s) sent to a given map task. Default ``batch_size`` is ``None``. compute: The compute strategy to use for the map operation. * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks. * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks. * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers. * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n, initial_size=initial)`` to use an autoscaling actor pool from ``m`` to ``n`` workers, with an initial size of ``initial``. batch_format: If ``"default"`` or ``"numpy"``, batches are ``Dict[str, numpy.ndarray]``. If ``"pandas"``, batches are ``pandas.DataFrame``. If ``"pyarrow"``, batches are ``pyarrow.Table``. If ``batch_format`` is set to ``None`` input block format will be used. zero_copy_batch: Whether ``fn`` should be provided zero-copy, read-only batches. If this is ``True`` and no copy is required for the ``batch_format`` conversion, the batch is a zero-copy, read-only view on data in Ray's object store, which can decrease memory utilization and improve performance. Setting this to ``False``, will make a copy of the *whole* batch, therefore allowing UDF to modify underlying data buffers (like tensors, binary arrays, etc) in place. It's recommended to copy only the data you need to modify instead of resorting to copying the whole batch. fn_args: Positional arguments to pass to ``fn`` after the first argument. These arguments are top-level arguments to the underlying Ray task. fn_kwargs: Keyword arguments to pass to ``fn``. These arguments are top-level arguments to the underlying Ray task. fn_constructor_args: Positional arguments to pass to ``fn``'s constructor. You can only provide this if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. fn_constructor_kwargs: Keyword arguments to pass to ``fn``'s constructor. This can only be provided if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. num_cpus: The number of CPUs to reserve for each parallel map worker. num_gpus: The number of GPUs to reserve for each parallel map worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel map worker. memory: The heap memory in bytes to reserve for each parallel map worker. concurrency: This argument is deprecated. Use ``compute`` argument. udf_modifying_row_count: Set to True if the UDF may modify the number of rows it receives so the limit pushdown optimization will not be applied. ray_remote_args_fn: A function that returns a dictionary of remote args passed to each map worker. The purpose of this argument is to generate dynamic arguments for each actor/task, and will be called each time prior to initializing the worker. Args returned from this dict will always override the args in ``ray_remote_args``. Note: this is an advanced, experimental feature. ray_remote_args: Additional resource requirements to request from Ray for each map worker. See :func:`ray.remote` for details. .. note:: The size of the batches provided to ``fn`` might be smaller than the specified ``batch_size`` if ``batch_size`` doesn't evenly divide the block(s) sent to a given map task. If ``batch_size`` is set and each input block is smaller than the ``batch_size``, Ray Data will bundle up many blocks as the input for one task, until their total size is equal to or greater than the given ``batch_size``. If ``batch_size`` is not set, the bundling will not be performed. Each task will receive entire input block as a batch. .. seealso:: :meth:`~Dataset.iter_batches` Call this function to iterate over batches of data. :meth:`~Dataset.take_batch` Call this function to get a batch of data from the dataset in the same format as will be passed to the `fn` function of :meth:`~Dataset.map_batches`. :meth:`~Dataset.flat_map` Call this method to create new records from existing ones. Unlike :meth:`~Dataset.map`, a function passed to :meth:`~Dataset.flat_map` can return multiple records. :meth:`~Dataset.map` Call this method to transform one record at time. Nrra&You must provide `batch_size` to `map_batches` when requesting GPUs. The optimal batch size depends on the model, data, and GPU used. We recommend using the largest batch size that doesn't result in your GPU device running out of memory. You can view the GPU memory usage via the Ray dashboard.z!Batch size can't be negative or 0r) ValueErrorrint*_map_batches_without_batch_size_validation)rrrrrrrrrrrrrrrrruse_gpuss r map_batcheszDataset.map_batchessR4'8HqL  +zY/F/F/  j# & & B:>>@AA A>t>  !%+ 3"7#$;1  !   rc |dkrtjdtd}t|||| }| | |d<| | |d<| | |d<t |}|t vrt dt d||j}t|j j ||||||||| |||| }t||j }t||S) Nrz|Passing 'default' to `map_batches` is deprecated and won't be supported after September 2025. Use `batch_size=None` instead.rrrrz The batch format must be one of , got: ) rrrmin_rows_per_bundled_inputrrrrrrrr)warningswarnDeprecationWarningrKr]rUrrrr6rrr,rr)rrrrrrrrrrrrrrrrrrmap_batches_oprs rrz2Dataset._map_batches_without_batch_size_validationsI4  " " MQ"    J&  3#      *2OJ '  *2OJ '  (.OH %*<88 2 2 2"3F""""  z  #   " !%+'1 3"7$;1+    #>4<@@ t\***ralpha)r stability column_nameexprc ddlm}ddlm}ddlm}|j}t||r6||j j |j g|g|}t||j } nN||j j t||g|} t| |j } t!|| S)a Add a new column to the dataset via an expression. This method allows you to add a new column to a dataset by applying an expression. The expression can be composed of existing columns, literals, and user-defined functions (UDFs). Examples: >>> import ray >>> from ray.data.expressions import col >>> ds = ray.data.range(100) >>> # Add a new column 'id_2' by multiplying 'id' by 2. >>> ds.with_column("id_2", col("id") * 2).show(2) {'id': 0, 'id_2': 0} {'id': 1, 'id_2': 2} >>> # Using a UDF with with_column >>> from ray.data.datatype import DataType >>> from ray.data.expressions import udf >>> import pyarrow.compute as pc >>> >>> @udf(return_dtype=DataType.int32()) ... def add_one(column): ... return pc.add(column, 1) >>> >>> ds.with_column("id_plus_one", add_one(col("id"))).show(2) {'id': 0, 'id_plus_one': 1} {'id': 1, 'id_plus_one': 2} Args: column_name: The name of the new column. expr: An expression that defines the new column values. **ray_remote_args: Additional resource requirements to request from Ray for the map tasks (e.g., `num_gpus=1`). Returns: A new dataset with the added column evaluated via the expression. r)r8)Download) DownloadExpr)uri_column_namesoutput_bytes_column_namesr)exprsr)1ray.data._internal.logical.operators.map_operatorr88ray.data._internal.logical.operators.one_to_one_operatorrray.data.expressionsrrrrrruri_column_namer,rrwaliasr) rrrrr8rrr download_opr project_ops r with_columnzDataset.with_column(s\ NMMMMMUUUUUU 655555z   dL ) ) A"("&"&"6!7+6- / K '{DLAALL "&zz4::k#:#:; /J 'z4<@@Lt\***rzUse `with_column` API instead)rpandas)rrrrxc gd}|vrtd|ddtdtffd }ts"td|j|f||dd |S) amAdd the given column to the dataset. A function generating the new column values given the batch in pyarrow or pandas format must be specified. This function must operate on batches of `batch_format`. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.schema() Column Type ------ ---- id int64 Add a new column equal to ``id * 2``. >>> ds.add_column("new_id", lambda df: df["id"] * 2).schema() Column Type ------ ---- id int64 new_id int64 Time complexity: O(dataset size / parallelism) Args: col: Name of the column to add. If the name already exists, the column is overwritten. fn: Map function generating the column values given a batch of records in pandas format. batch_format: If ``"default"`` or ``"numpy"``, batches are ``Dict[str, numpy.ndarray]``. If ``"pandas"``, batches are ``pandas.DataFrame``. If ``"pyarrow"``, batches are ``pyarrow.Table``. If ``"numpy"``, batches are ``Dict[str, numpy.ndarray]``. compute: This argument is deprecated. Use ``concurrency`` argument. concurrency: The maximum number of Ray workers to use concurrently. ray_remote_args: Additional resource requirements to request from Ray (e.g., num_gpus=1 to request GPUs for the map tasks). See :func:`ray.remote` for details. )rpyarrownumpyz$batch_format argument must be on of rbatchrc:|}dkrBddl}t||j|j|jfr |j|_||jddf<|Sdkrddl}t||j|j fsJdt||j }|dkr| |S|||St|tjsJdt|||<|S)NrrrzIFor pyarrow batch format, the function must return a pyarrow Array, got: zGFor numpy batch format, the function must return a numpy.ndarray, got: )rr DataFrameIndexSeriesindexlocrArray ChunkedArrayrschemaget_field_index append_column set_columnnpndarray)rcolumnpdpa column_idxrrxrs r add_columnz&Dataset.add_column..add_columnsfRYYFx''####fr|RXry&IJJ/#(;FL$* !!!S&! **$$$$!&28R_*EFF2#'<<22F#\99#>> ## ..sF;;;'' C@@@ "&"*55:+/<<::5$c  rz`fn` must be callable, got {}T)rrrr)rrXcallableformatr) rrxrrrrraccepted_batch_formatsrs ``` rrzDataset.add_columnnst"@!?!? 5 5 5'7M''$''  & i& I& & & & & & & & P|| I<CCBGGHH Ht  %#       r)rrcolsc tttkrtdfd}|j|fdd||d|S)aDrop one or more columns from the dataset. Examples: >>> import ray >>> ds = ray.data.read_parquet("s3://anonymous@ray-example-data/iris.parquet") >>> ds.schema() Column Type ------ ---- sepal.length double sepal.width double petal.length double petal.width double variety string >>> ds.drop_columns(["variety"]).schema() Column Type ------ ---- sepal.length double sepal.width double petal.length double petal.width double Time complexity: O(dataset size / parallelism) Args: cols: Names of the columns to drop. If any name does not exist, an exception is raised. Column names must be unique. compute: This argument is deprecated. Use ``concurrency`` argument. concurrency: The maximum number of Ray workers to use concurrently. ray_remote_args: Additional resource requirements to request from Ray (e.g., num_gpus=1 to request GPUs for the map tasks). See :func:`ray.remote` for details. z/drop_columns expects unique column names, got: c.|Sr)drop)rrs r drop_columnsz*Dataset.drop_columns..drop_columnss::d## #rrT)rrrr)lensetrr)rrrrrrs ` rrzDataset.drop_columnssV t99CII & &UtUUVV V $ $ $ $ $ t  " #       rc b ddlm t|tr  |g}nt|trvt d|Dst dt|tt|krt d| fd|D}ntdddl m }|| }|j }t|jj||| }t#||j} t'|| S) aSelect one or more columns from the dataset. Specified columns must be in the dataset schema. .. tip:: If you're reading parquet files with :meth:`ray.data.read_parquet`, you might be able to speed it up by using projection pushdown; see :ref:`Parquet column pruning ` for details. Examples: >>> import ray >>> ds = ray.data.read_parquet("s3://anonymous@ray-example-data/iris.parquet") >>> ds.schema() Column Type ------ ---- sepal.length double sepal.width double petal.length double petal.width double variety string >>> ds.select_columns(["sepal.length", "sepal.width"]).schema() Column Type ------ ---- sepal.length double sepal.width double Time complexity: O(dataset size / parallelism) Args: cols: Names of the columns to select. If a name isn't in the dataset schema, an exception is raised. Columns also should be unique. compute: This argument is deprecated. Use ``concurrency`` argument. concurrency: The maximum number of Ray workers to use concurrently. ray_remote_args: Additional resource requirements to request from Ray (e.g., num_gpus=1 to request GPUs for the map tasks). See :func:`ray.remote` for details. r)rxc3@K|]}t|tVdSrrstr.0rxs r z)Dataset.select_columns..Qs,<<z#s++<<<<<z*Dataset.select_columns..Zs!***SSVV***rzCselect_columns requires 'cols' to be a string or a list of strings.TaskPoolStrategysizerrr)rrxrr listallrrr TypeErrorray.data._internal.computerrrr8rrr,rr) rrrrrrrr select_oprrxs @rselect_columnszDataset.select_columnssx^ -,,,,, dC  SYYKEE d # # <>> import ray >>> ds = ray.data.read_parquet("s3://anonymous@ray-example-data/iris.parquet") >>> ds.schema() Column Type ------ ---- sepal.length double sepal.width double petal.length double petal.width double variety string You can pass a dictionary mapping old column names to new column names. >>> ds.rename_columns({"variety": "category"}).schema() Column Type ------ ---- sepal.length double sepal.width double petal.length double petal.width double category string Or you can pass a list of new column names. >>> ds.rename_columns( ... ["sepal_length", "sepal_width", "petal_length", "petal_width", "variety"] ... ).schema() Column Type ------ ---- sepal_length double sepal_width double petal_length double petal_width double variety string Args: names: A dictionary that maps old column names to new column names, or a list of new column names. concurrency: The maximum number of Ray workers to use concurrently. ray_remote_args: Additional resource requirements to request from Ray (e.g., num_gpus=1 to request GPUs for the map tasks). See :func:`ray.remote` for details. z0rename_columns received 'names' with no entries.z9rename_columns received duplicate values in the 'names': c3pK|]1\}}t|tot|tV2dSrr)r kvs rr z)Dataset.rename_columns..sP>Ba 1c""9z!S'9'9rzJrename_columns requires both keys and values in the 'names' to be strings.cXg|]'\}}t||(Srrx_renamer prevnews rrz*Dataset.rename_columns..s0KKK cSYY&&s++KKKrz>rename_columns requires 'names' with at least one column name.c3@K|]}t|tVdSrrr s rr z)Dataset.rename_columns..s,==z#s++======rzBrename_columns requires all elements in the 'names' to be strings.z!rename_columns requires 'names': z$ length match current schema names: .cXg|]'\}}t||(Srr"r$s rrz*Dataset.rename_columns..s0WWW cSYY&&s++WWWrzLrename_columns expected names to be either List[str] or Dict[str, str], got Nz;Expected `concurrency` to be an integer or `None`, but got rrrr)rdictrrvaluesrritemsrrrziprrrrrrrr8rrrwr,rr) rrrrr current_namesrrrrrs rrename_columnszDataset.rename_columnsnsp eT " "/  U !STTT5<<>>""c#ellnn*=*=&>&>>> WPUWWFKkkmm !% LKU[[]]KKKEE t $ $   T5zzSU__,, WPUWW==u=====  X!KKMM/M=!!SZZ// 666%2666 XWSPU=V=VWWWEE6'+E{{666   ":k3+G+G "&"&&&  @?????"" 444z     "::&&+    #9dl;; t\***rc t|||| }t||| | } |j} t |jj||||||| |  }t||j}t| |S)aApply the given function to each row and then flatten results. Use this method if your transformation returns multiple rows for each input row. You can use either a function or a callable class to perform the transformation. For functions, Ray Data uses stateless Ray tasks. For classes, Ray Data uses stateful Ray actors. For more information, see :ref:`Stateful Transforms `. .. tip:: :meth:`~Dataset.map_batches` can also modify the number of rows. If your transformation is vectorized like most NumPy and pandas operations, it might be faster. .. warning:: Specifying both ``num_cpus`` and ``num_gpus`` for map tasks is experimental, and may result in scheduling or stability issues. Please `report any issues `_ to the Ray team. Examples: .. testcode:: from typing import Any, Dict, List import ray def duplicate_row(row: Dict[str, Any]) -> List[Dict[str, Any]]: return [row] * 2 print( ray.data.range(3) .flat_map(duplicate_row) .take_all() ) .. testoutput:: [{'id': 0}, {'id': 0}, {'id': 1}, {'id': 1}, {'id': 2}, {'id': 2}] Time complexity: O(dataset size / parallelism) Args: fn: The function or generator to apply to each record, or a class type that can be instantiated to create such a callable. compute: The compute strategy to use for the map operation. * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks. * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks. * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers. * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n, initial_size=initial)`` to use an autoscaling actor pool from ``m`` to ``n`` workers, with an initial size of ``initial``. fn_args: Positional arguments to pass to ``fn`` after the first argument. These arguments are top-level arguments to the underlying Ray task. fn_kwargs: Keyword arguments to pass to ``fn``. These arguments are top-level arguments to the underlying Ray task. fn_constructor_args: Positional arguments to pass to ``fn``'s constructor. You can only provide this if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. fn_constructor_kwargs: Keyword arguments to pass to ``fn``'s constructor. This can only be provided if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. num_cpus: The number of CPUs to reserve for each parallel map worker. num_gpus: The number of GPUs to reserve for each parallel map worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel map worker. memory: The heap memory in bytes to reserve for each parallel map worker. concurrency: This argument is deprecated. Use ``compute`` argument. ray_remote_args_fn: A function that returns a dictionary of remote args passed to each map worker. The purpose of this argument is to generate dynamic arguments for each actor/task, and will be called each time prior to initializing the worker. Args returned from this dict will always override the args in ``ray_remote_args``. Note: this is an advanced, experimental feature. ray_remote_args: Additional resource requirements to request from Ray for each map worker. See :func:`ray.remote` for details. .. seealso:: :meth:`~Dataset.map_batches` Call this method to transform batches of data. :meth:`~Dataset.map` Call this method to transform one row at time. r) input_oprrrrrrrr) rKrLrrr5rrr,rr)rrrrrrrrrrrrrroprs rflat_mapzDataset.flat_maps`'  3#    =       z   '+ 3"71+    #2t|44 t\***rc t|du|dug}|dkrtdfd}t|| | | } |jj}d}d}d}d}d}d}d}|m|dddlm}t|tr8tj dtd dd l m }||}n|}|| }netj d t|s%tdt!|jd|}}}}}t%||| }t'||||||||| |  }|j}t-||j}t1||S)axFilter out rows that don't satisfy the given predicate. You can use either a function or a callable class or an expression to perform the transformation. For functions, Ray Data uses stateless Ray tasks. For classes, Ray Data uses stateful Ray actors. For more information, see :ref:`Stateful Transforms `. .. tip:: If you use the `expr` parameter with a predicate expression, Ray Data optimizes your filter with native Arrow interfaces. .. deprecated:: String expressions are deprecated and will be removed in a future version. Use predicate expressions from `ray.data.expressions` instead. Examples: >>> import ray >>> from ray.data.expressions import col >>> ds = ray.data.range(100) >>> # String expressions (deprecated - will warn) >>> ds.filter(expr="id <= 4").take_all() [{'id': 0}, {'id': 1}, {'id': 2}, {'id': 3}, {'id': 4}] >>> # Using predicate expressions (preferred) >>> ds.filter(expr=(col("id") > 10) & (col("id") < 20)).take_all() [{'id': 11}, {'id': 12}, {'id': 13}, {'id': 14}, {'id': 15}, {'id': 16}, {'id': 17}, {'id': 18}, {'id': 19}] Time complexity: O(dataset size / parallelism) Args: fn: The predicate to apply to each row, or a class type that can be instantiated to create such a callable. expr: An expression that represents a predicate (boolean condition) for filtering. Can be either a string expression (deprecated) or a predicate expression from `ray.data.expressions`. fn_args: Positional arguments to pass to ``fn`` after the first argument. These arguments are top-level arguments to the underlying Ray task. fn_kwargs: Keyword arguments to pass to ``fn``. These arguments are top-level arguments to the underlying Ray task. fn_constructor_args: Positional arguments to pass to ``fn``'s constructor. You can only provide this if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. fn_constructor_kwargs: Keyword arguments to pass to ``fn``'s constructor. This can only be provided if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. compute: The compute strategy to use for the map operation. * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks. * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks. * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers. * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n, initial_size=initial)`` to use an autoscaling actor pool from ``m`` to ``n`` workers, with an initial size of ``initial``. num_cpus: The number of CPUs to reserve for each parallel map worker. num_gpus: The number of GPUs to reserve for each parallel map worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel map worker. memory: The heap memory in bytes to reserve for each parallel map worker. concurrency: This argument is deprecated. Use ``compute`` argument. ray_remote_args_fn: A function that returns a dictionary of remote args passed to each map worker. The purpose of this argument is to generate dynamic arguments for each actor/task, and will be called each time prior to initializing the worker. Args returned from this dict will always override the args in ``ray_remote_args``. Note: this is an advanced, experimental feature. ray_remote_args: Additional resource requirements to request from Ray (e.g., num_gpus=1 to request GPUs for the map tasks). See :func:`ray.remote` for details. Nrz/Exactly one of 'fn' or 'expr' must be provided.c>td|ddS)Nzwhen 'z]' is used, 'fn_args/fn_kwargs' or 'fn_constructor_args/fn_constructor_kwargs' cannot be used.r) param_typerrrrs r_check_fn_params_incompatiblez5Dataset.filter.._check_fn_params_incompatiblesO#(&2(4 GZGGG54rrrrzString expressions are deprecated and will be removed in a future version. Use predicate expressions from ray.data.expressions instead. For example: from ray.data.expressions import col; ds.filter(expr=col('column_name') > 5)) stacklevel)ExpressionEvaluatorrz@Use 'expr' instead of 'fn' when possible for performant filters.z*fn must be a UserDefinedFunction, but got z instead.)rrrr) r1predicate_exprrrrrrrrr)sumrrLrrrrrr rrr?ray.data._internal.planner.plan_expression.expression_evaluatorr;parse_native_expressionrr__name__rKr4rrr,rr)rrrrrrrrrrrrrrprovided_paramsr8r1r< filter_fnfilter_fn_argsfilter_fn_kwargsfilter_fn_constructor_argsfilter_fn_constructor_kwargsfilter_computerr; filter_oprrs ```` rfilterzDataset.filterysi~r~t4/?@AA a  NOO O        =       %))-37 2659>B"AE$48   ) )& 1 1 1 C C C C C C$$$ & =' "5!L!LT!R!R"&--;???NN MR   B<<  4Bxx(444 I$N( )< &+@ (1$7' N)"& :">"1+    z  "9dl;; t\***r)shufflekeyssort num_blockstarget_num_rows_per_blockrJrKrLc|D|tjd|durtjd|rtjd||td||td||rtd|j}|t |jj| }nt|jj|||| }t||j }t||S) aDRepartition the :class:`Dataset` into exactly this number of :ref:`blocks `. This method can be useful to tune the performance of your pipeline. To learn more, see :ref:`Advanced: Performance Tips and Tuning `. If you're writing data to files, you can also use this method to change the number of output files. To learn more, see :ref:`Changing the number of output files `. .. note:: Repartition has three modes: * When ``num_blocks`` and ``shuffle=True`` are specified Ray Data performs a full distributed shuffle producing exactly ``num_blocks`` blocks. * When ``num_blocks`` and ``shuffle=False`` are specified, Ray Data does NOT perform full shuffle, instead opting in for splitting and combining of the blocks attempting to minimize the necessary data movement (relative to full-blown shuffle). Exactly ``num_blocks`` will be produced. * If ``target_num_rows_per_block`` is set (exclusive with ``num_blocks`` and ``shuffle``), streaming repartitioning will be executed, where blocks will be made to carry no more than ``target_num_rows_per_block`` rows. Smaller blocks will be combined into bigger ones up to ``target_num_rows_per_block`` as well. .. image:: /data/images/dataset-shuffle.svg :align: center .. https://docs.google.com/drawings/d/132jhE3KXZsf29ho1yUdPrCHB9uheHBWHJhDQMXqIVPA/edit Examples: >>> import ray >>> ds = ray.data.range(100).repartition(10).materialize() >>> ds.num_blocks() 10 Time complexity: O(dataset size / parallelism) Args: num_blocks: Number of blocks after repartitioning. target_num_rows_per_block: [Experimental] The target number of rows per block to repartition. Performs streaming repartitioning of the dataset (no shuffling). Note that either `num_blocks` or `target_num_rows_per_block` must be set, but not both. When `target_num_rows_per_block` is set, it only repartitions :class:`Dataset` :ref:`blocks ` that are larger than `target_num_rows_per_block`. Note that the system will internally figure out the number of rows per :ref:`blocks ` for optimal execution, based on the `target_num_rows_per_block`. This is the current behavior because of the implementation and may change in the future. shuffle: Whether to perform a distributed shuffle during the repartition. When shuffle is enabled, each output block contains a subset of data rows from each input block, which requires all-to-all data movement. When shuffle is disabled, output blocks are created from adjacent input blocks, minimizing data movement. keys: List of key columns repartitioning will use to determine which partition will row belong to after repartitioning (by applying hash-partitioning algorithm to the whole dataset). Note that, this config is only relevant when `DataContext.use_hash_based_shuffle` is set to True. sort: Whether the blocks should be sorted after repartitioning. Note, that by default blocks will be sorted in the ascending order. Note that you must set either `num_blocks` or `target_num_rows_per_block` but not both. Additionally note that this operation materializes the entire dataset in memory when you set shuffle to True. Returns: The repartitioned :class:`Dataset`. Nz:`keys` is ignored when `target_num_rows_per_block` is set.Fz:`sort` is ignored when `target_num_rows_per_block` is set.z=`shuffle` is ignored when `target_num_rows_per_block` is set.z>Either `num_blocks` or `target_num_rows_per_block` must be setzROnly one of `num_blocks` or `target_num_rows_per_block` must be set, but not both.z@`shuffle` must be False when `target_num_rows_per_block` is set.)rN) num_outputsrJrKrL) rrrrrr9rrr/r,rr) rrMrNrJrKrLrr2rs r repartitionzDataset.repartition@s`\ % 0 P5   P  S  %>%FP   ")B)N   % 0W 0R z   $ 0%"&*CBB "&& B#2t|44 t\***r)seedrMrRc |td|j}t|jj||}t ||j}t||S)ajRandomly shuffle the rows of this :class:`Dataset`. .. tip:: This method can be slow. For better performance, try :ref:`Iterating over batches with shuffling `. Also, see :ref:`Optimizing shuffles `. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.random_shuffle().take(3) # doctest: +SKIP {'id': 41}, {'id': 21}, {'id': 92}] >>> ds.random_shuffle(seed=42).take(3) # doctest: +SKIP {'id': 77}, {'id': 21}, {'id': 63}] Time complexity: O(dataset size / parallelism) Args: seed: Fix the random seed to use, otherwise one is chosen based on system randomness. Returns: The shuffled :class:`Dataset`. Nz`num_blocks` parameter is deprecated in Ray 2.9. random_shuffle() does not support to change the number of output blocks. Use repartition() instead.)rRr) rrrr.rrr,rr)rrRrMrrr2rs rrandom_shufflezDataset.random_shuffles{F  !$)  z      "+    #2t|44 t\***rrRc|j}t|jj|}t ||j}t||S)a,Randomly shuffle the :ref:`blocks ` of this :class:`Dataset`. This method is useful if you :meth:`~Dataset.split` your dataset into shards and want to randomize the data in each shard without performing a full :meth:`~Dataset.random_shuffle`. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.take(5) [{'id': 0}, {'id': 1}, {'id': 2}, {'id': 3}, {'id': 4}] >>> ds.randomize_block_order().take(5) # doctest: +SKIP {'id': 15}, {'id': 16}, {'id': 17}, {'id': 18}, {'id': 19}] Args: seed: Fix the random seed to use, otherwise one is chosen based on system randomness. Returns: The block-shuffled :class:`Dataset`. rU)rrr-rrr,rr)rrRrr2rs rrandomize_block_orderzDataset.randomize_block_ordersX:z      "   #2t|44 t\***rfractionc,ddlddl|jdkrt ddksdkrt dddlmdtdttffd }| ||gdd S) amReturns a new :class:`Dataset` containing a random fraction of the rows. .. note:: This method returns roughly ``fraction * total_rows`` rows. An exact number of rows isn't guaranteed. Examples: >>> import ray >>> ds1 = ray.data.range(100) >>> ds1.random_sample(0.1).count() # doctest: +SKIP 10 >>> ds2 = ray.data.range(1000) >>> ds2.random_sample(0.123, seed=42).take(2) # doctest: +SKIP [{'id': 2}, {'id': 9}] >>> ds2.random_sample(0.123, seed=42).take(2) # doctest: +SKIP [{'id': 2}, {'id': 9}] Args: fraction: The fraction of elements to sample. seed: Seeds the python random pRNG generator. Returns: Returns a :class:`Dataset` containing the sampled rows. rNz$Cannot sample from an empty Dataset.rz!Fraction must be between 0 and 1. TaskContextrrRcj}d|jvr|jd}n[|)tj}||jd<n0tj|j|g}||jd<tj|t|kd}t|j r| |St|j r|j |ddfStdt|)NrngrzUnsupported batch type: ) get_currentkwargsrrandom default_rngtask_idxwhererrTabletakerilocrr) rrRctxr]mask_idxr[rXrrs r random_samplez,Dataset.random_sample..random_sample<s))++C ""j'i++--$' 5!!i++S\4,@AA$' 5!x 3u:: 6 6 ABB1EH%** /zz(+++E2<00 /z(AAA+..EU EEFF Fr)rrr) rrrinitial_num_blocksr4ray.data._internal.execution.interfaces.task_contextr[rXr rr)rrXrRrir[rrs ` @@@rrizDataset.random_samples:  : ( ( * *a / /CDD D a<<8a<<@AA ATTTTTT G G(3- G G G G G G G G G( F     r)equallocality_hintsnrlrmrrc|j}t|jj|||}t ||j}t||}||j tj |||S)a Returns ``n`` :class:`DataIterators ` that can be used to read disjoint subsets of the dataset in parallel. This method is the recommended way to consume :class:`Datasets ` for distributed training. Streaming split works by delegating the execution of this :class:`Dataset` to a coordinator actor. The coordinator pulls block references from the executed stream, and divides those blocks among ``n`` output iterators. Iterators pull blocks from the coordinator actor to return to their caller on ``next``. The returned iterators are also repeatable; each iteration will trigger a new execution of the Dataset. There is an implicit barrier at the start of each iteration, which means that `next` must be called on all iterators before the iteration starts. .. warning:: Because iterators are pulling blocks from the same :class:`Dataset` execution, if one iterator falls behind, other iterators may be stalled. Examples: .. testcode:: import ray ds = ray.data.range(100) it1, it2 = ds.streaming_split(2, equal=True) Consume data from iterators in parallel. .. testcode:: @ray.remote def consume(it): for batch in it.iter_batches(): pass ray.get([consume.remote(it1), consume.remote(it2)]) You can loop over the iterators multiple times (multiple epochs). .. testcode:: @ray.remote def train(it): NUM_EPOCHS = 2 for _ in range(NUM_EPOCHS): for batch in it.iter_batches(): pass ray.get([train.remote(it1), train.remote(it2)]) The following remote function call blocks waiting for a read on ``it2`` to start. .. testcode:: :skipif: True ray.get(train.remote(it1)) Args: n: Number of output iterators to return. equal: If ``True``, each output iterator sees an exactly equal number of rows, dropping data if necessary. If ``False``, some iterators may see slightly more or less rows than others, but no data is dropped. locality_hints: Specify the node ids corresponding to each iterator location. Dataset will try to minimize data movement based on the iterator output locations. This list must have length ``n``. You can get the current node id of a task or actor by calling ``ray.get_runtime_context().get_node_id()``. Returns: The output iterator splits. These iterators are Ray-serializable and can be freely passed to any Ray task or actor. .. seealso:: :meth:`Dataset.split` Unlike :meth:`~Dataset.streaming_split`, :meth:`~Dataset.split` materializes the dataset in memory. ) num_splitsrlrm) rrr<rrr,rrr_uuidr+create)rrnrlrmrr2r split_datasets rstreaming_splitzDataset.streaming_splitWsxz      ")     #2t|44 l33  +++&-mQOOOrMaterializedDatasetc 2  !"#dkrtdd|rY|W|}|z##fdtddzD}||}|dS|r6t |kr#tdt |dd |j d "|j}t j \}} |tj |} tj | } g} t| | D]\} } "fd t| |D}tt| |j}| t!t#||j|| St't|| !d t(dt*t,dt.t,t(fffd }dt*t0t2dt.t4t*t0t2ffd}dt*t,dt.t,t4ffd}|t ||}||}||}t7jt:}|D]r}||}||}||}g}|rOt ||kr<|||rt ||k<|||<st;t>j !|"}|D]m}t ||||krL|||t ||||kLnt |dksJt |g}|D]_}||}!fd|D} tGtIt|| " j% | `|rtM|"}g} |D]n tt g |j}| t!t#||j|o| S)aMaterialize and split the dataset into ``n`` disjoint pieces. This method returns a list of ``MaterializedDataset`` that can be passed to Ray Tasks and Actors and used to read the dataset rows in parallel. Examples: .. testcode:: @ray.remote class Worker: def train(self, data_iterator): for batch in data_iterator.iter_batches(batch_size=8): pass workers = [Worker.remote() for _ in range(4)] shards = ray.data.range(100).split(n=4, equal=True) ray.get([w.train.remote(s) for w, s in zip(workers, shards)]) Time complexity: O(1) Args: n: Number of child datasets to return. equal: Whether to guarantee each split has an equal number of records. This might drop records if the rows can't be divided equally among the splits. locality_hints: [Experimental] A list of Ray actor handles of size ``n``. The system tries to co-locate the blocks of the i-th dataset with the i-th actor to maximize data locality. Returns: A list of ``n`` disjoint dataset splits. .. seealso:: :meth:`Dataset.split_at_indices` Unlike :meth:`~Dataset.split`, which splits a dataset into approximately equal splits, :meth:`Dataset.split_proportionately` lets you split a dataset into different sizes. :meth:`Dataset.split_proportionately` This method is equivalent to :meth:`Dataset.split_at_indices` if you compute indices manually. :meth:`Dataset.streaming_split`. Unlike :meth:`~Dataset.split`, :meth:`~Dataset.streaming_split` doesn't materialize the dataset in memory. rzThe number of splits z is not positive.Ncg|]}|zSrr)r i split_indexs rrz!Dataset.split..sFFF[1_FFFrrzThe length of locality_hints z$ doesn't equal the number of splits r(FcJg|]\}}t||fgj S) owns_blocksrr&r)r bmbundleowned_by_consumers rrz!Dataset.split..sN1Q.? r input_datarMactorsrct|}||z}||zz }i}t|D] \}}|||<||kr||xxdz cc<!|S)zGiven the total number of blocks and a list of actors, calcuate the expected number of blocks to allocate for each actor. r)r enumerate) rMr num_actorsnum_blocks_per_actornum_blocks_leftnum_blocks_by_actorrxactorrns rbuild_allocation_size_mapz0Dataset.split..build_allocation_size_map=s VJ#-#; (+?!+CCO"$ %f-- 4 45-A#E*&&'...!3...& &rblocksc"tj|}tjt }|D]S}||idg}|r|dnd}|||T|S)zBuild the reverse index from node_id to block_refs. For simplicity, if the block is stored on multiple nodes we only pick the first one. node_idsrN)ray experimentalget_object_locations collections defaultdictrgetappend)rblock_ref_locationsblock_refs_by_node_id block_refrnode_ids rbuild_block_refs_by_node_idz2Dataset.split..build_block_refs_by_node_idMs#&"2"G"G"O"O $/$;D$A$A !# A A .229bAAEEjRTUU)1;(1++t%g.55i@@@@( (rcftjjfd|DS)z(Build a map from a actor to its node_id.ci|]W}||jididXS)AddressNodeID)r _actor_idhex)r r actors_states r zADataset.split..build_node_id_by_actor.._se|''(;(;(=(=rBBY##Xr)r_privatestater)rrs @rbuild_node_id_by_actorz-Dataset.split..build_node_id_by_actor\sF<-4466L$  rc g|] }| Srr)r r~metadata_mappings rrz!Dataset.split..s<<<(+<<->-@-@AA$ " !J 9 9::" ' '%)#Y ' #s(^ ' ' ' ' ' ' )5)* ) #tIe,-- . ) ) ) ) 49 c3h    )B(A  OO^) ) %!< ;J G G11.AA*6t<< $ 5 5E&u-G3G>> import ray >>> ds = ray.data.range(10) >>> d1, d2, d3 = ds.split_at_indices([2, 5]) >>> d1.take_batch() {'id': array([0, 1])} >>> d2.take_batch() {'id': array([2, 3, 4])} >>> d3.take_batch() {'id': array([5, 6, 7, 8, 9])} Time complexity: O(num splits) Args: indices: List of sorted integers which indicate where the dataset are split. If an index exceeds the length of the dataset, an empty dataset is returned. Returns: The dataset splits. .. seealso:: :meth:`Dataset.split` Unlike :meth:`~Dataset.split_at_indices`, which lets you split a dataset into different sizes, :meth:`Dataset.split` splits a dataset into approximately equal splits. :meth:`Dataset.split_proportionately` This method is equivalent to :meth:`Dataset.split_at_indices` if you compute indices manually. :meth:`Dataset.streaming_split`. Unlike :meth:`~Dataset.split`, :meth:`~Dataset.streaming_split` doesn't materialize the dataset in memory. rz$indices must be at least of length 1zindices must be sortedrzindices must be positiveFSplitrparentcJg|]\}}t||fgdj S)Fr{r})r r~rrs rrz,Dataset.split_at_indices..sDAqAq6(fmLLLrr)rrsortedtime perf_counterrrrDrrr-rE time_total_sr,r2rrrur@r)rr start_timerrsplit_duration parent_statssplitsbsmsrrrrs @rrzDataset.split_at_indicessT w<>788 8&((  J..00, M     *,,z9z'')) &(++  FB 7B- MMME!/E BKKK'[111 L MM#!%):):)<)<==       r proportionsct|dkrtdt|dkrtdtd|Drtd|t j|}fd|D}d}tt|dz d d D]<}||xx|zcc<||||dzkr|dz }||xxdzcc<=td |Drtd ||S) aMaterialize and split the dataset using proportions. A common use case for this is splitting the dataset into train and test sets (equivalent to eg. scikit-learn's ``train_test_split``). For a higher level abstraction, see :meth:`Dataset.train_test_split`. This method splits datasets so that all splits always contains at least one row. If that isn't possible, an exception is raised. This is equivalent to caulculating the indices manually and calling :meth:`Dataset.split_at_indices`. Examples: >>> import ray >>> ds = ray.data.range(10) >>> d1, d2, d3 = ds.split_proportionately([0.2, 0.5]) >>> d1.take_batch() {'id': array([0, 1])} >>> d2.take_batch() {'id': array([2, 3, 4, 5, 6])} >>> d3.take_batch() {'id': array([7, 8, 9])} Time complexity: O(num splits) Args: proportions: List of proportions to split the dataset according to. Must sum up to less than 1, and each proportion must be bigger than 0. Returns: The dataset splits. .. seealso:: :meth:`Dataset.split` Unlike :meth:`~Dataset.split_proportionately`, which lets you split a dataset into different sizes, :meth:`Dataset.split` splits a dataset into approximately equal splits. :meth:`Dataset.split_at_indices` :meth:`Dataset.split_proportionately` uses this method under the hood. :meth:`Dataset.streaming_split`. Unlike :meth:`~Dataset.split`, :meth:`~Dataset.streaming_split` doesn't materialize the dataset in memory. rz(proportions must be at least of length 1z#proportions must sum to less than 1c3"K|] }|dkV dSrNr)r ps rr z0Dataset.split_proportionately..- s&++!qAv++++++rz!proportions must be bigger than 0c4g|]}t|zSr)r)r  proportiondataset_lengths rrz1Dataset.split_proportionately..2 s3   1;C+ , ,   rrr9rc3"K|] }|dkV dSrr)r rxs rr z0Dataset.split_proportionately..= s&--!qAv------rz>> import ray >>> ds = ray.data.range(8) >>> train, test = ds.train_test_split(test_size=0.25) >>> train.take_batch() {'id': array([0, 1, 2, 3, 4, 5])} >>> test.take_batch() {'id': array([6, 7])} Args: test_size: If float, should be between 0.0 and 1.0 and represent the proportion of the dataset to include in the test split. If int, represents the absolute number of test samples. The train split always complements the test split. shuffle: Whether or not to globally shuffle the dataset before splitting. Defaults to ``False``. This may be a very expensive operation with a large dataset. seed: Fix the random seed to use for shuffle, otherwise one is chosen based on system randomness. Ignored if ``shuffle=False``. stratify: Optional column name to use for stratified sampling. If provided, the splits will maintain the same proportions of each class in the stratify column across both train and test sets. Returns: Train and test subsets as two ``MaterializedDatasets``. .. seealso:: :meth:`Dataset.split_proportionately` rUz%`test_size` must be int or float got r(NzCannot specify both 'shuffle=True' and 'stratify' parameters. Stratified splitting maintains class proportions and is incompatible with shuffling.r) rTrrfloatrrr_stratified_train_test_split_validate_test_size_floatr_validate_test_size_intrr)rrrJrRrr ds_lengths rtrain_test_splitzDataset.train_test_splitD s(V  ."""--B)c5\22 XVDOOVVVWW W  x+g   44RHMM M i ' ' @  * *9 5 5 5++Q]O<< <  ( (B 7 7 7 I&& I(='>?? ?rcttr||}|z n|fd}|||}|dtg}|dtg}||fS)a7Perform stratified train-test split on the dataset. Args: ds: The dataset to split. test_size: Test size as int or float. stratify: Column name to use for stratified sampling. Returns: Train and test subsets as two MaterializedDatasets. ct|}t|z}tjdg||z zdg|zz|t<|S)NTF)rrrarray_TRAIN_TEST_SPLIT_COLUMN) group_batchrn test_countrs radd_train_flagz.add_train_flag sZK  AQ]++J46H!j.)UGj,@@55K0 1 rc|tSrrrows rz6Dataset._stratified_train_test_split.. s 45rc|t Srrrs rrz6Dataset._stratified_train_test_split.. sC 899r) rrrrgroupby map_groups materializerIrr) rrrrrrsplit_dstrain_dstest_dss ` rrz$Dataset._stratified_train_test_split s i % % 644YCCI!I-II  * *9 5 5 5     ::h''22>BBNNPP?? 5 5  ,01 2 2 // 9 9  ,01 2 2   rcD|dks|dkrtd|ddS)zValidate test_size when it's a float. Args: test_size: Test size as float between 0 and 1. Raises: ValueError: If test_size is not in valid range. rrzLIf `test_size` is a float, it must be bigger than 0 and smaller than 1. Got r(Nr6)rrs rrz!Dataset._validate_test_size_float sE >>Y!^^,(,,, ,^rcr|}|dks||krtd|d|d|S)a1Validate test_size when it's an int and return dataset length. Args: test_size: Test size as int. ds: Dataset to validate against. Returns: Dataset length for reuse. Raises: ValueError: If test_size is not in valid range. rz]If `test_size` is an int, it must be bigger than 0 and smaller than the size of the dataset (z). Got r()rr)rrrrs rrzDataset._validate_test_size_int sbHHJJ >>Y)33$1:$$ $$$  r)rrr`) split_type hash_columnrRr)hashr`r)rrc  ddl ddl ddlm dksdkrt d|dkrt d|dkrt d d jf fd }d jd t j jff fd }|dkr|j|fddi|}n:|dkr"t d|j|fddi|}nt d||td tg} |td tg} | | fS)asplit the dataset into train and test subsets in a streaming manner. This method is recommended for large datasets. The split type can be either "hash" or "random". - "random": The dataset is split into random train and test subsets. - "hash": The dataset is split into train and test subsets based on the hash of the key column. .. tip:: Make sure to set the `preserve_order` flag in the `ExecutionOptions` to True to ensure that the split is deterministic across pipeline executions. This is important to avoid test rows to end up in the train set and vice versa on multiple executions. This can be set with ``ray.data.DataContext.get_current().execution_options.preserve_order = True``. Examples: Examples with Random split: >>> import ray >>> ctx = ray.data.DataContext.get_current() >>> ctx.execution_options.preserve_order = True >>> ds = ray.data.range(8) >>> train, test = ds.streaming_train_test_split(test_size=0.25, seed=0) >>> train.count() 6 >>> test.count() 2 >>> ctx.execution_options.preserve_order = False Examples with Hash split: >>> import ray >>> ds = ray.data.range(8) >>> train, test = ds.streaming_train_test_split(test_size=0.25, split_type="hash", hash_column="id") >>> train.take_batch() {'id': array([0, 2, 3, 4, 5, 6])} >>> test.take_batch() {'id': array([1, 7])} Args: test_size: The proportion of the dataset to include in the test split. Must be between 0.0 and 1.0. split_type: The type of split to perform. Can be "hash" or "random". hash_column: The column to use for the hash split. Required for hash split and ignored for random split. seed: The seed to use for the random split. Ignored for hash split. **ray_remote_kwargs: Additional kwargs to pass to the Ray remote function. Returns: Train and test subsets as two ``Dataset``. .. seealso:: :meth:`Dataset.train_test_split` rNrZrz"test_size must be between 0 and 1.rz$seed is not supported for hash splitr`z-hash_column is not supported for random splitrc}d|jvr|jd}nb0tj|jg}||jd<n0tj|jg}||jd<||jdz k}|t | S)a] Perform a random split on a batch: each row goes to train with probability (1 - test_proportion), or to test otherwise. This version ensures that the random choices are **stable per Ray task execution** by seeding the RNG with a combination of a user-specified seed and the Ray task ID. train_test_split_rngNrr) r^r_rr`rarbnum_rowsrrrbool_)rrgr]is_trainr[rrRrs r random_splitz8Dataset.streaming_train_test_split..random_split% s))++C%33j!78i++S\N;;58 122i++S\4,@AA58 12zz%.11Q]CH&&("((8"((**(*M*M rrc>dtdtffd |jvr|}nt ddfd|D}|t|S)Nkeyrctt|dd}|dz dzkrdndS)N) digest_sizebigrlTF)r from_bytesblake2br encodedigest)rhhashlibrs r key_to_bucketzMDataset.streaming_train_test_split..hash_split..key_to_bucket> sgNNOOCHHOO$5$51OEELLNNPU !A Mg#>>>ttEIrz Key column z not found in batchc&g|] }|Srr)r rrs rrzJDataset.streaming_train_test_split..hash_split..K s#"F"F"F#==#5#5"F"F"Frr ) rr column_namesto_numpyrrr rr)rrK bucket_arrrrrrrs @r hash_splitz6Dataset.streaming_train_test_split..hash_split= s J3 J3 J J J J J J Je000[)2244 !O{!O!O!OPPP"F"F"F"F"F"F"FRXXZZXXJ&&'?LL Lrrrz&hash_column is required for hash splitzInvalid split type: z == True)rz == False) rrrkr[rrdrrrIrr)rrrrrRray_remote_kwargsr r buckettedds_trainds_testr[rrs ` `` @@@rstreaming_train_test_splitz"Dataset.streaming_train_test_split sG~ TTTTTT >>Y!^^ABB B   f 4 4CDD D  "zX'='=LMM M          0 Mbh M5281C+D M M M M M M M M M"  ! !((&$II 6 ! !" !IJJJ((&$II @J@@AA A##,666$  ,01 2 2 "",777#  ,01 2 2   rotherctj}|gt|z}d|D}td|D}t ||j}t dgid|D}tj|z |_tt||j |S)aConcatenate :class:`Datasets ` across rows. The order of the blocks in the datasets is preserved, as is the relative ordering between the datasets passed in the argument list. .. caution:: Unioned datasets aren't lineage-serializable. As a result, they can't be used as a tunable hyperparameter in Ray Tune. Examples: >>> import ray >>> ds1 = ray.data.range(2) >>> ds2 = ray.data.range(3) >>> ds1.union(ds2).take_all() [{'id': 0}, {'id': 1}, {'id': 0}, {'id': 1}, {'id': 2}] Args: other: List of datasets to combine with this one. The datasets must have the same schema as this dataset, otherwise the behavior is undefined. Returns: A new dataset holding the rows of the input datasets. c&g|]}|jjSr)rr)r union_dss rrz!Dataset.union.. sOOO(5OOOrcg|] }|j Sr)r)r rs rrz!Dataset.union.. s 1 1 14dh 1 1 1rrc@g|]}|jSr)rr)r ds rrz!Dataset.union.. s"666AGMMOO666rr) rrrUnionLogicalOperatorr,rrErrr@r)rr%rdatasets logical_plansr2rrs runionz Dataset.unionh s6&(( 6DKK'OOhOOO ! 1 1= 1 1 1 #2t|44 r]66X666   ".00:= %!2!2!4!4 5 5    rid)partition_size_hintaggregator_ray_remote_argsvalidate_schemas join_typenum_partitionsonright_on left_suffix right_suffixr2r3r4c ft|ttfs%tdt |jd|rAt|ttfs%tdt |jd|p|}| r?|} |} tj| | |||j } t|j j |j j ||||||||  }t| t||jS)aJoin :class:`Datasets ` on join keys. Args: ds: Other dataset to join against join_type: The kind of join that should be performed, one of ("inner", "left_outer", "right_outer", "full_outer", "left_semi", "right_semi", "left_anti", "right_anti"). num_partitions: Total number of "partitions" input sequences will be split into with each partition being joined independently. Increasing number of partitions allows to reduce individual partition size, hence reducing memory requirements when individual partitions are being joined. Note that, consequently, this will also be a total number of blocks that will be produced as a result of executing join. on: The columns from the left operand that will be used as keys for the join operation. right_on: The columns from the right operand that will be used as keys for the join operation. When none, `on` will be assumed to be a list of columns to be used for the right dataset as well. left_suffix: (Optional) Suffix to be appended for columns of the left operand. right_suffix: (Optional) Suffix to be appended for columns of the right operand. partition_size_hint: (Optional) Hint to joining operator about the estimated avg expected size of the individual partition (in bytes). This is used in estimating the total dataset size and allow to tune memory requirement of the individual joining workers to prevent OOMs when joining very large datasets. aggregator_ray_remote_args: (Optional) Parameter overriding `ray.remote` args passed when constructing joining (aggregator) workers. validate_schemas: (Optional) Controls whether validation of provided configuration against input schemas will be performed (defaults to false, since obtaining schemas could be prohibitively expensive). Returns: A :class:`Dataset` that holds rows of input left Dataset joined with the right Dataset based on join type and keys. Examples: .. testcode:: :skipif: True doubles_ds = ray.data.range(4).map( lambda row: {"id": row["id"], "double": int(row["id"]) * 2} ) squares_ds = ray.data.range(4).map( lambda row: {"id": row["id"], "square": int(row["id"]) ** 2} ) # Inner join example joined_ds = doubles_ds.join( squares_ds, join_type="inner", num_partitions=2, on=("id",), ) print(sorted(joined_ds.take_all(), key=lambda item: item["id"])) .. testoutput:: :options: +ELLIPSIS, +NORMALIZE_WHITESPACE [ {'id': 0, 'double': 0, 'square': 0}, {'id': 1, 'double': 2, 'square': 1}, {'id': 2, 'double': 4, 'square': 4}, {'id': 3, 'double': 6, 'square': 9} ] .. testcode:: :skipif: True # Left anti-join example: find rows in doubles_ds that don't match squares_ds partial_squares_ds = ray.data.range(2).map( lambda row: {"id": row["id"] + 2, "square": int(row["id"]) ** 2} ) anti_joined_ds = doubles_ds.join( partial_squares_ds, join_type="left_anti", num_partitions=2, on=("id",), ) print(sorted(anti_joined_ds.take_all(), key=lambda item: item["id"])) .. testoutput:: :options: +ELLIPSIS, +NORMALIZE_WHITESPACE [ {'id': 0, 'double': 0}, {'id': 1, 'double': 2} ] .. testcode:: :skipif: True # Left semi-join example: find rows in doubles_ds that have matches in squares_ds # (only returns columns from left dataset) semi_joined_ds = doubles_ds.join( squares_ds, join_type="left_semi", num_partitions=2, on=("id",), ) print(sorted(semi_joined_ds.take_all(), key=lambda item: item["id"])) .. testoutput:: :options: +ELLIPSIS, +NORMALIZE_WHITESPACE [ {'id': 0, 'double': 0}, {'id': 1, 'double': 2}, {'id': 2, 'double': 4}, {'id': 3, 'double': 6} ] z$Expected tuple or list as `on` (got )z*Expected tuple or list as `right_on` (got ) left_input_opright_input_opleft_key_columnsright_key_columnsr5r6left_columns_suffixright_columns_suffixr2r3)rrrrrr@rr3_validate_schemasrrrrrr,r)rrr5r6r7r8r9r:r2r3r4left_op_schemaright_op_schemarr2s rjoinz Dataset.join s@R"udm,, KtBxx7HKKK   Jx%?? WT(^^=TWWW  >r  R15N24))++O  ">?B Q Q Qz   ,0+/&) +!- 3'A    t[T\::;;;rrrtcddlm}|6t||d||dkrt d||||S)aGroup rows of a :class:`Dataset` according to a column. Use this method to transform data based on a categorical variable. Examples: .. testcode:: import pandas as pd import ray def normalize_variety(group: pd.DataFrame) -> pd.DataFrame: for feature in group.drop(columns=["variety"]).columns: group[feature] = group[feature] / group[feature].abs().max() return group ds = ( ray.data.read_parquet("s3://anonymous@ray-example-data/iris.parquet") .groupby("variety") .map_groups(normalize_variety, batch_format="pandas") ) Time complexity: O(dataset size * log(dataset size / parallelism)) Args: key: A column name or list of column names. If this is ``None``, place all rows in a single group. num_partitions: Number of partitions data will be partitioned into (only relevant if hash-shuffling strategy is used). When not set defaults to `DataContext.min_parallelism`. Returns: A lazy :class:`~ray.data.grouped_data.GroupedData`. .. seealso:: :meth:`~ray.data.grouped_data.GroupedData.map_groups` Call this method to transform groups of data. rrsNFfetch_if_missingz+`num_partitions` must be a positive integer)r6)ray.data.grouped_datartrAvalidate_schemarr)rrr6rts rrzDataset.groupbyF s~` 655555 ? CLL ( (e)L)L M M M  %.A*=*=JKK K{4^DDDDrr ignore_nullscf|t||}||S)aList the unique elements in a given column. Examples: >>> import ray >>> ds = ray.data.from_items([1, 2, 3, 2, 3]) >>> sorted(ds.unique("item")) [1, 2, 3] This function is very useful for computing labels in a machine learning dataset: >>> import ray >>> ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") >>> sorted(ds.unique("target")) [0, 1, 2] One common use case is to convert the class labels into integers for training and inference: >>> classes = {0: 'Setosa', 1: 'Versicolor', 2: 'Virginica'} >>> def preprocessor(df, classes): ... df["variety"] = df["target"].map(classes) ... return df >>> train_ds = ds.map_batches( ... preprocessor, fn_kwargs={"classes": classes}, batch_format="pandas") >>> train_ds.sort("sepal length (cm)").take(1) # Sort to make it deterministic [{'sepal length (cm)': 4.3, ..., 'variety': 'Setosa'}] Time complexity: O(dataset size / parallelism) Args: column: The column to collect unique elements over. ignore_nulls: If ``True``, ignore null values in the column. Returns: A list with unique elements in the given column. rL) _aggregate_onrT_aggregate_result)rrrLrets runiquezDataset.unique s3T  l KK%%c***raggsc|dj|d}t|dkr|dndS)arAggregate values using one or more functions. Use this method to compute metrics like the product of a column. Examples: .. testcode:: import ray from ray.data.aggregate import AggregateFn ds = ray.data.from_items([{"number": i} for i in range(1, 10)]) aggregation = AggregateFn( init=lambda column: 1, # Apply this to each row to produce a partial aggregate result accumulate_row=lambda a, row: a * row["number"], # Apply this to merge partial aggregate results into a final result merge=lambda a1, a2: a1 * a2, name="prod" ) print(ds.aggregate(aggregation)) .. testoutput:: {'prod': 362880} Time complexity: O(dataset size / parallelism) Args: *aggs: :class:`Aggregations ` to perform. Returns: A ``dict`` where each each value is an aggregation for a given column. Nrr)r aggregaterer)rrSrQs rrUzDataset.aggregate sJL+dll4  *D166q99SAs1vv4/rcf|t||}||S)auCompute the sum of one or more columns. Examples: >>> import ray >>> ray.data.range(100).sum("id") 4950 >>> ray.data.from_items([ ... {"A": i, "B": i**2} ... for i in range(100) ... ]).sum(["A", "B"]) {'sum(A)': 4950, 'sum(B)': 328350} Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values are ignored when computing the sum. If ``False``, when a null value is encountered, the output is ``None``. Ray Data considers ``np.nan``, ``None``, and ``pd.NaT`` to be null values. Default is ``True``. Returns: The sum result. For different values of ``on``, the return varies: - ``on=None``: a dict containing the column-wise sum of all columns, - ``on="col"``: a scalar representing the sum of all items in column ``"col"``, - ``on=["col_1", ..., "col_n"]``: an n-column ``dict`` containing the column-wise sum of the provided columns. If the dataset is empty, all values are null. If ``ignore_nulls`` is ``False`` and any value is null, then the output is ``None``. rN)rOrSrPrr7rLrQs rr=z Dataset.sum 3R  b| DD%%c***rcf|t||}||S)amReturn the minimum of one or more columns. Examples: >>> import ray >>> ray.data.range(100).min("id") 0 >>> ray.data.from_items([ ... {"A": i, "B": i**2} ... for i in range(100) ... ]).min(["A", "B"]) {'min(A)': 0, 'min(B)': 0} Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values are ignored when computing the min; if ``False``, when a null value is encountered, the output is ``None``. This method considers ``np.nan``, ``None``, and ``pd.NaT`` to be null values. Default is ``True``. Returns: The min result. For different values of ``on``, the return varies: - ``on=None``: an dict containing the column-wise min of all columns, - ``on="col"``: a scalar representing the min of all items in column ``"col"``, - ``on=["col_1", ..., "col_n"]``: an n-column dict containing the column-wise min of the provided columns. If the dataset is empty, all values are null. If ``ignore_nulls`` is ``False`` and any value is null, then the output is ``None``. rN)rOrQrPrWs rminz Dataset.min rXrcf|t||}||S)arReturn the maximum of one or more columns. Examples: >>> import ray >>> ray.data.range(100).max("id") 99 >>> ray.data.from_items([ ... {"A": i, "B": i**2} ... for i in range(100) ... ]).max(["A", "B"]) {'max(A)': 99, 'max(B)': 9801} Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values are ignored when computing the max; if ``False``, when a null value is encountered, the output is ``None``. This method considers ``np.nan``, ``None``, and ``pd.NaT`` to be null values. Default is ``True``. Returns: The max result. For different values of ``on``, the return varies: - ``on=None``: an dict containing the column-wise max of all columns, - ``on="col"``: a scalar representing the max of all items in column ``"col"``, - ``on=["col_1", ..., "col_n"]``: an n-column dict containing the column-wise max of the provided columns. If the dataset is empty, all values are null. If ``ignore_nulls`` is ``False`` and any value is null, then the output is ``None``. rN)rOrOrPrWs rmaxz Dataset.max2 rXrcf|t||}||S)aCompute the mean of one or more columns. Examples: >>> import ray >>> ray.data.range(100).mean("id") 49.5 >>> ray.data.from_items([ ... {"A": i, "B": i**2} ... for i in range(100) ... ]).mean(["A", "B"]) {'mean(A)': 49.5, 'mean(B)': 3283.5} Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values are ignored when computing the mean; if ``False``, when a null value is encountered, the output is ``None``. This method considers ``np.nan``, ``None``, and ``pd.NaT`` to be null values. Default is ``True``. Returns: The mean result. For different values of ``on``, the return varies: - ``on=None``: an dict containing the column-wise mean of all columns, - ``on="col"``: a scalar representing the mean of all items in column ``"col"``, - ``on=["col_1", ..., "col_n"]``: an n-column dict containing the column-wise mean of the provided columns. If the dataset is empty, all values are null. If ``ignore_nulls`` is ``False`` and any value is null, then the output is ``None``. rN)rOrPrPrWs rmeanz Dataset.mean^ s3R  r  EE%%c***rrddofch|t|||}||S)aCompute the standard deviation of one or more columns. .. note:: This method uses Welford's online method for an accumulator-style computation of the standard deviation. This method has numerical stability, and is computable in a single pass. This may give different (but more accurate) results than NumPy, Pandas, and sklearn, which use a less numerically stable two-pass algorithm. To learn more, see `the Wikapedia article `_. Examples: >>> import ray >>> round(ray.data.range(100).std("id", ddof=0), 5) 28.86607 >>> result = ray.data.from_items([ ... {"A": i, "B": i**2} ... for i in range(100) ... ]).std(["A", "B"]) >>> [(key, round(value, 10)) for key, value in result.items()] [('std(A)', 29.0114919759), ('std(B)', 2968.1748039269)] Args: on: a column name or a list of column names to aggregate. ddof: Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. ignore_nulls: Whether to ignore null values. If ``True``, null values are ignored when computing the std; if ``False``, when a null value is encountered, the output is ``None``. This method considers ``np.nan``, ``None``, and ``pd.NaT`` to be null values. Default is ``True``. Returns: The standard deviation result. For different values of ``on``, the return varies: - ``on=None``: an dict containing the column-wise std of all columns, - ``on="col"``: a scalar representing the std of all items in column ``"col"``, - ``on=["col_1", ..., "col_n"]``: an n-column dict containing the column-wise std of the provided columns. If the dataset is empty, all values are null. If ``ignore_nulls`` is ``False`` and any value is null, then the output is ``None``. )rLr_)rOrRrP)rr7r_rLrQs rstdz Dataset.std s5p  b|$ OO%%c***rcolumnsoverride_dtype_agg_mappingrucddlm}m}m}m}||||}|jstd||ndd|dj |j}| dd} |j } |j } || | | |j\} } }|| || d }|| || d }|||| t| S) aGenerate a statistical summary of the dataset, organized by data type. This method computes various statistics for different column dtypes: - For numerical dtypes (int*, float*, decimal, bool): count, mean, min, max, std, approx_quantile (median), missing%, zero% - For string and binary dtypes: count, missing%, approx_top_k (top 10 values) - For temporal dtypes (timestamp, date, time, duration): count, min, max, missing% - For other dtypes: count, missing%, approx_top_k You can customize the aggregations performed for specific data types using the `override_dtype_agg_mapping` parameter. The summary separates statistics into two tables: - Schema-matching stats: Statistics that preserve the original column type (e.g., min/max for integers) - Schema-changing stats: Statistics that change the type (e.g., mean converts int to float) Examples: >>> import ray >>> ds = ray.data.from_items([ ... {"age": 25, "salary": 50000, "name": "Alice", "city": "NYC"}, ... {"age": 30, "salary": 60000, "name": None, "city": "LA"}, ... {"age": 0, "salary": None, "name": "Bob", "city": None}, ... ]) >>> summary = ds.summary() >>> # Get combined pandas DataFrame with all statistics >>> summary.to_pandas() # doctest: +SKIP statistic age city name salary 0 approx_quantile[0] 25.000000 None None 60000.000000 1 approx_topk[0] NaN {'city': 'LA', 'count': 1} {'count': 1, 'name': 'Bob'} NaN 2 approx_topk[1] NaN {'city': 'NYC', 'count': 1} {'count': 1, 'name': 'Alice'} NaN 3 count 3.000000 3 3 3.000000 4 max 30.000000 NaN NaN 60000.000000 5 mean 18.333333 None None 55000.000000 6 min 0.000000 NaN NaN 50000.000000 7 missing_pct 0.000000 33.333333 33.333333 33.333333 8 std 13.123346 None None 5000.000000 9 zero_pct 33.333333 None None 0.000000 >>> # Access individual column statistics >>> summary.get_column_stats("age") # doctest: +SKIP statistic value 0 approx_quantile[0] 25.000000 1 approx_topk[0] NaN 2 approx_topk[1] NaN 3 count 3.000000 4 max 30.000000 5 mean 18.333333 6 min 0.000000 7 missing_pct 0.000000 8 std 13.123346 9 zero_pct 33.333333 Custom aggregations for specific types: >>> from ray.data.datatype import DataType >>> from ray.data.aggregate import Sum, Count >>> # Override aggregations for int64 columns >>> custom_mapping = { ... DataType.int64(): lambda col: [Count(on=col), Sum(on=col)] ... } >>> summary = ds.summary(override_dtype_agg_mapping=custom_mapping) Args: columns: Optional list of column names to include in the summary. If None, all columns will be included. override_dtype_agg_mapping: Optional mapping from DataType to factory functions. Each factory function takes a column name and returns a list of aggregators for that column. This will be merged with the default mapping, with user-provided mappings taking precedence. Returns: A DatasetSummary object with methods to access statistics and the original dataset schema. Use `to_pandas()` to get all statistics as a DataFrame, or `get_column_stats(col)` for a specific column r)ru_build_summary_table_dtype_aggregators_for_dataset_parse_summary_stats)rbdtype_agg_mappingzPsummary() requires at least one column with a supported type. Columns provided: Nrz. Check that the specified columns exist and have supported types (numeric, string, binary, or temporal). Columns with None or object types are skipped.rT)preserve_typesF)_stats_matching_column_dtype_stats_mismatching_column_dtypedataset_schemarb) ray.data.statsrurerfrgr aggregatorsrrrUre base_schemar)rrbrcrurerfrg dtype_aggs aggs_dataset agg_resultoriginal_schema agg_schemaschema_matching_statsschema_changing_stats all_columnsschema_matching_tableschema_changing_tables rsummaryzDataset.summary sj            43 KKMM8   % ,070CWW,,, 4t||D))3Z5KL !&&q))!, ++--3!((**6 ! Z5K   ! !  !5 4 !;PT! ! ! !5 4 !;PU! ! ! ~)>,A*%%     r descending boundariesc|tdt|||}|j}t |jj|}t||j}t||S)avSort the dataset by the specified key column or key function. The `key` parameter must be specified (i.e., it cannot be `None`). .. note:: If provided, the `boundaries` parameter can only be used to partition the first sort key. Examples: >>> import ray >>> ds = ray.data.range(15) >>> ds = ds.sort("id", descending=False, boundaries=[5, 10]) >>> for df in ray.get(ds.to_pandas_refs()): ... print(df) id 0 0 1 1 2 2 3 3 4 4 id 0 5 1 6 2 7 3 8 4 9 id 0 10 1 11 2 12 3 13 4 14 Time complexity: O(dataset size * log(dataset size / parallelism)) Args: key: The column or a list of columns to sort by. descending: Whether to sort in descending order. Must be a boolean or a list of booleans matching the number of the columns. boundaries: The list of values based on which to repartition the dataset. For example, if the input boundary is [10,20], rows with values less than 10 will be divided into the first block, rows with values greater than or equal to 10 and less than 20 will be divided into the second block, and rows with values greater than or equal to 20 will be divided into the third block. If not provided, the boundaries will be sampled from the input blocks. This feature only supports numeric columns right now. Returns: A new, sorted :class:`Dataset`. Raises: ``ValueError``: if the sort key is None. Nz/The 'key' parameter cannot be None for sorting.)sort_key) rrArrr0rrr,rr)rrr{r|r~rr2rs rrLz Dataset.sortN s~z ;NOO O3 J77z      "   #2t|44 t\***rc|j}t|jjgd|DR}t ||j}t||S)aZip the columns of this dataset with the columns of another. The datasets must have the same number of rows. Their column sets are merged, and any duplicate column names are disambiguated with suffixes like ``"_1"``. .. note:: The smaller of the two datasets is repartitioned to align the number of rows per block with the larger dataset. .. note:: Zipped datasets aren't lineage-serializable. As a result, they can't be used as a tunable hyperparameter in Ray Tune. Examples: >>> import ray >>> ds1 = ray.data.range(5) >>> ds2 = ray.data.range(5) >>> ds3 = ray.data.range(5) >>> ds1.zip(ds2, ds3).take_batch() {'id': array([0, 1, 2, 3, 4]), 'id_1': array([0, 1, 2, 3, 4]), 'id_2': array([0, 1, 2, 3, 4])} Args: *other: List of datasets to combine with this one. The datasets must have the same row count as this dataset, otherwise the ValueError is raised. Returns: A :class:`Dataset` containing the columns of the second dataset concatenated horizontally with the columns of the first dataset, with duplicate column names disambiguated with suffixes like ``"_1"``. Raises: ValueError: If the datasets have different row counts. c&g|]}|jjSr)rr)r r%s rrzDataset.zip.. s*V*V*Vu5+>+B*V*V*Vr)rrr:rrr,rr)rr%rr2rs rr-z Dataset.zip sbJz   #' W*V*VPU*V*V*V W W W"2t|44 t\***rlimitc|j}t|jj|}t ||j}t||S)a5Truncate the dataset to the first ``limit`` rows. Unlike :meth:`~Dataset.take`, this method doesn't move data to the caller's machine. Instead, it returns a new :class:`Dataset` pointing to the truncated distributed data. Examples: >>> import ray >>> ds = ray.data.range(1000) >>> ds.limit(5).count() 5 Time complexity: O(limit specified) Args: limit: The size of the dataset to truncate to. Returns: The truncated dataset. )r)rrr;rrr,rr)rrrr2rs rrz Dataset.limit sN,z   4%) 7 7 7"2t|44 t\***r)rc ^t|}||} tt||d|}n#t $rt dwxYw||j |j_ |S)aReturn up to ``batch_size`` rows from the :class:`Dataset` in a batch. Ray Data represents batches as NumPy arrays or pandas DataFrames. You can configure the batch type by specifying ``batch_format``. This method is useful for inspecting inputs to :meth:`~Dataset.map_batches`. .. warning:: :meth:`~Dataset.take_batch` moves up to ``batch_size`` rows to the caller's machine. If ``batch_size`` is large, this method can cause an ` ``OutOfMemory`` error on the caller. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.take_batch(5) {'id': array([0, 1, 2, 3, 4])} Time complexity: O(batch_size specified) Args: batch_size: The maximum number of rows to return. batch_format: If ``"default"`` or ``"numpy"``, batches are ``Dict[str, numpy.ndarray]``. If ``"pandas"``, batches are ``pandas.DataFrame``. Returns: A batch of up to ``batch_size`` rows from the dataset. Raises: ``ValueError``: if the dataset is empty. r)rprefetch_batchesrzThe dataset is empty.) r]rnextiter iter_batches StopIterationr_synchronize_progress_barrr_snapshot_stats)rrr limited_dsress r take_batchzDataset.take_batch sN+<88 ZZ ++  6++#-)*%1,CC 6 6 6455 5 6 &&(((&0%5%;%;%=%= " s 2AA3ctjdrtdg}||}|D],}||t||krn-| |j |j _ |S)a$Return up to ``limit`` rows from the :class:`Dataset`. This method is useful for inspecting data. .. warning:: :meth:`~Dataset.take` moves up to ``limit`` rows to the caller's machine. If ``limit`` is large, this method can cause an ``OutOfMemory`` error on the caller. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.take(3) [{'id': 0}, {'id': 1}, {'id': 2}] Time complexity: O(limit specified) Args: limit: The maximum number of rows to return. Returns: A list of up to ``limit`` rows from the dataset. .. seealso:: :meth:`~Dataset.take_all` Call this method to return all rows. dataset_takezgTip: Use `take_batch()` instead of `take() / show()` to return records in pandas or numpy batch format.) rutillog_onceloggerinfor iter_rowsrrrrrr)rroutputrrs rrez Dataset.takesB 8  ^ , ,  KK;   ZZ&& ''))  C MM#   6{{e##$ &&(((&0%5%;%;%=%= " rcg}|D]?}|||&t||krtd|d@||S)aTReturn all of the rows in this :class:`Dataset`. This method is useful for inspecting small datasets. .. warning:: :meth:`~Dataset.take_all` moves the entire dataset to the caller's machine. If the dataset is large, this method can cause an ``OutOfMemory`` error on the caller. Examples: >>> import ray >>> ds = ray.data.range(5) >>> ds.take_all() [{'id': 0}, {'id': 1}, {'id': 2}, {'id': 3}, {'id': 4}] Time complexity: O(dataset size) Args: limit: Raise an error if the size exceeds the specified limit. Returns: A list of all the rows in the dataset. .. seealso:: :meth:`~Dataset.take` Call this method to return a specific number of rows. Nz-The dataset has more than the given limit of z records.)rrrrr)rrrrs rtake_allzDataset.take_allJs@>>##  C MM#    S[[5%8%8 TETTT &&((( rcT||D]}t|dS)a)Print up to the given number of rows from the :class:`Dataset`. This method is useful for inspecting data. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.show(3) {'id': 0} {'id': 1} {'id': 2} Time complexity: O(limit specified) Args: limit: The maximum number of row to print. .. seealso:: :meth:`~Dataset.take` Call this method to get (not print) a given number of rows. N)reprint)rrrs rshowz Dataset.showts6499U##  C #JJJJ  rz row countz Examples:)if_more_than_readdatasource_metadatapatternc*|jdkrdS|}||S|j}t t |jjg}t||j }t||}d}| dD]O}tj |vsJdtj d||tj  z }Pt|S)aCount the number of rows in the dataset. For Datasets which only read Parquet files (created with :meth:`~ray.data.read_parquet`), this method reads the file metadata to efficiently count the number of rows without reading in the entire data. Examples: >>> import ray >>> ds = ray.data.range(10) >>> ds.count() 10 Returns: The number of records in the dataset. rN)r)rzIOutputs from the 'Count' logical operator should contain a column named '')rrj _meta_countrr1r8rrr,rrr COLUMN_NAMEr=r)r meta_countrcount_oprcount_dsrrs rrz Dataset.counts. : ( ( * *a / /1%%''  ! z  !3!7rBBBCC"8T\:: 4..**d*;; 4 4E$---/+///.-- U5,-1133 3EE5zzrrz-or if ``fetch_if_missing=True`` (the default)zTime complexity:)rrextra_conditionrrISchemac0|jj}|jd}|t||S|dj|}|+|j|t||SdS)avReturn the schema of the dataset. Examples: >>> import ray >>> ds = ray.data.range(10) >>> ds.schema() Column Type ------ ---- id int64 Time complexity: O(1) Args: fetch_if_missing: If True, synchronously fetch the schema if it's not known. If False, None is returned if the schema is not known. Default is True. Returns: The :class:`ray.data.Schema` class of the records, or None if the schema is not known and fetch_if_missing is False. FrHN data_contextr)r_contextrrr cache_schema)rrIrros rrzDataset.schemas<*%j'''??  "+G<<< < jjmm)00BR0SS  " J # #K 0 0 0+G<<< <4rcD||}||jSdS)aReturns the columns of this Dataset. Time complexity: O(1) Example: >>> import ray >>> # Create dataset from synthetic data. >>> ds = ray.data.range(1000) >>> ds.columns() ['id'] Args: fetch_if_missing: If True, synchronously fetch the column names from the schema if it's not known. If False, None is returned if the schema is not known. Default is True. Returns: A list of the column names for this Dataset or None if schema is not known and `fetch_if_missing` is False. rHN)rr)rrIrs rrbzDataset.columnss+:.>??  < trc td)aReturn the number of blocks of this :class:`Dataset`. This method is only implemented for :class:`~ray.data.MaterializedDataset`, since the number of blocks may dynamically change during execution. For instance, during read and transform operations, Ray Data may dynamically adjust the number of blocks to respect memory limits, increasing the number of blocks at runtime. Returns: The number of blocks of this :class:`Dataset`. zNumber of blocks is only available for `MaterializedDataset`,because the number of blocks may dynamically change during execution.Call `ds.materialize()` to get a `MaterializedDataset`.)NotImplementedErrorrs rrMzDataset.num_blockss" F   rc|jjj#|jjjS|jj}|r |djdStd|DS)a9Return the in-memory size of the dataset. Examples: >>> import ray >>> ds = ray.data.range(10) >>> ds.size_bytes() 80 Returns: The in-memory size of the dataset in bytes, or None if the in-memory size is not known. Nrc3$K|] }|jV dSr) size_bytes)r rs rr z%Dataset.size_bytes..=s$22A1<222222r)rrinfer_metadatarrrrr=)rrs rrzDataset.size_bytes's   ! 0 0 2 2 = I%)88::E E:%%''0 8A;19422222222rchtt|jS)aReturn the list of input files for the dataset. Examples: >>> import ray >>> ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") >>> ds.input_files() ['ray-example-data/iris.csv'] Returns: The list of input files used to create the dataset, or an empty list if the input files is not known. )rrr input_filesrs rrzDataset.input_files?s(C ..0011222r) partition_cols filesystemtry_create_dirarrow_open_stream_argsfilename_providerarrow_parquet_args_fnmin_rows_per_filemax_rows_per_filerrnum_rows_per_filemodepathrrzpyarrow.fs.FileSystemrrrrrrrrrc  |d}t| || \}}t|||||||||||j|  }||| | dS)akWrites the :class:`~ray.data.Dataset` to parquet files under the provided ``path``. The number of files is determined by the number of blocks in the dataset. To control the number of number of blocks, call :meth:`~ray.data.Dataset.repartition`. If pyarrow can't represent your data, this method errors. By default, the format of the output files is ``{uuid}_{block_idx}.parquet``, where ``uuid`` is a unique id for the dataset. To modify this behavior, implement a custom :class:`~ray.data.datasource.FilenameProvider` and pass it in as the ``filename_provider`` argument. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.write_parquet("local:///tmp/data/") Time complexity: O(dataset size / parallelism) Args: path: The path to the destination root directory, where parquet files are written to. partition_cols: Column names by which to partition the dataset. Files are writted in Hive partition style. filesystem: The pyarrow filesystem implementation to write to. These filesystems are specified in the `pyarrow docs `_. Specify this 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. try_create_dir: If ``True``, attempts to create all directories in the destination path. Does nothing if all directories already exist. Defaults to ``True``. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_output_stream `_, which is used when opening the file to write to. filename_provider: A :class:`~ray.data.datasource.FilenameProvider` implementation. Use this parameter to customize what your filenames look like. The filename is expected to be templatized with `{i}` to ensure unique filenames when writing multiple files. If it's not templatized, Ray Data will add `{i}` to the filename to ensure compatibility with the pyarrow `write_dataset `_. arrow_parquet_args_fn: Callable that returns a dictionary of write arguments that are provided to `pyarrow.parquet.ParquetWriter() `_ when writing each block to a file. Overrides any duplicate keys from ``arrow_parquet_args``. If `row_group_size` is provided, it will be passed to `pyarrow.parquet.ParquetWriter.write_table() `_. Use this argument instead of ``arrow_parquet_args`` if any of your write arguments can't pickled, or if you'd like to lazily resolve the write arguments for each dataset block. min_rows_per_file: [Experimental] The target minimum number of rows to write to each file. If ``None``, Ray Data writes a system-chosen number of rows to each file. If the number of rows per block is larger than the specified value, Ray Data writes the number of rows per block to each file. The specified value is a hint, not a strict limit. Ray Data might write more or fewer rows to each file. max_rows_per_file: [Experimental] The target maximum number of rows to write to each file. If ``None``, Ray Data writes a system-chosen number of rows to each file. If the number of rows per block is smaller than the specified value, Ray Data writes the number of rows per block to each file. The specified value is a hint, not a strict limit. Ray Data might write more or fewer rows to each file. If both ``min_rows_per_file`` and ``max_rows_per_file`` are specified, ``max_rows_per_file`` takes precedence when they cannot both be satisfied. ray_remote_args: Kwargs passed to :func:`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. num_rows_per_file: [Deprecated] Use min_rows_per_file instead. arrow_parquet_args: Options to pass to `pyarrow.parquet.ParquetWriter() `_, which is used to write out each block to a file. See `arrow_parquet_args_fn` for more detail. mode: Determines how to handle existing files. Valid modes are "overwrite", "error", "ignore", "append". Defaults to "append". NOTE: This method isn't atomic. "Overwrite" first deletes all the data before writing to `path`. NciSrrrrrrz'Dataset.write_parquet..sBr)rrr) rrarrow_parquet_argsrrrropen_stream_argsr dataset_uuidrrr)rJr!rqwrite_datasink)rrrrrrrrrrrrrrreffective_min_rowseffective_max_rowsdatasinks r write_parquetzDataset.write_parquetPs^ ! ($.J !1M///2 2 2 .. # )"7100!)3/      +#      r) rrrrpandas_json_args_fnrrrrrrc |d}t| |\} }t||| | |||||j|  }|||| dS)a#Writes the :class:`~ray.data.Dataset` to JSON and JSONL files. The number of files is determined by the number of blocks in the dataset. To control the number of number of blocks, call :meth:`~ray.data.Dataset.repartition`. This method is only supported for datasets with records that are convertible to pandas dataframes. By default, the format of the output files is ``{uuid}_{block_idx}.json``, where ``uuid`` is a unique id for the dataset. To modify this behavior, implement a custom :class:`~ray.data.datasource.FilenameProvider` and pass it in as the ``filename_provider`` argument. Examples: Write the dataset as JSON file to a local directory. >>> import ray >>> import pandas as pd >>> ds = ray.data.from_pandas([pd.DataFrame({"one": [1], "two": ["a"]})]) >>> ds.write_json("local:///tmp/data") Write the dataset as JSONL files to a local directory. >>> ds = ray.data.read_json("s3://anonymous@ray-example-data/train.jsonl") >>> ds.write_json("local:///tmp/data") Time complexity: O(dataset size / parallelism) Args: path: The path to the destination root directory, where the JSON files are written to. filesystem: The pyarrow filesystem implementation to write to. These filesystems are specified in the `pyarrow docs `_. Specify this 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. try_create_dir: If ``True``, attempts to create all directories in the destination path. Does nothing if all directories already exist. Defaults to ``True``. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_output_stream `_, which is used when opening the file to write to. filename_provider: A :class:`~ray.data.datasource.FilenameProvider` implementation. Use this parameter to customize what your filenames look like. pandas_json_args_fn: Callable that returns a dictionary of write arguments that are provided to `pandas.DataFrame.to_json() `_ when writing each block to a file. Overrides any duplicate keys from ``pandas_json_args``. Use this parameter instead of ``pandas_json_args`` if any of your write arguments can't be pickled, or if you'd like to lazily resolve the write arguments for each dataset block. min_rows_per_file: [Experimental] The target minimum number of rows to write to each file. If ``None``, Ray Data writes a system-chosen number of rows to each file. If the number of rows per block is larger than the specified value, Ray Data writes the number of rows per block to each file. The specified value is a hint, not a strict limit. Ray Data might write more or fewer rows to each file. ray_remote_args: kwargs passed to :func:`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. num_rows_per_file: Deprecated. Use ``min_rows_per_file`` instead. pandas_json_args: These args are passed to `pandas.DataFrame.to_json() `_, which is used under the hood to write out each :class:`~ray.data.Dataset` block. These are dict(orient="records", lines=True) by default. mode: Determines how to handle existing files. Valid modes are "overwrite", "error", "ignore", "append". Defaults to "append". NOTE: This method isn't atomic. "Overwrite" first deletes all the data before writing to `path`. NciSrrrrrrz$Dataset.write_json..Bs"rrr) rpandas_json_argsrrrrrrrr)rJrrqr)rrrrrrrrrrrrrr_rs r write_jsonzDataset.write_jsonsJ  &",*  </CT! ! ! A  3-0!)3/      +#      rtable_identifiercatalog_kwargssnapshot_propertiesrboverwrite_filterrv upsert_kwargsoverwrite_kwargsc bt|||||||} || || dS)aWrites the :class:`~ray.data.Dataset` to an Iceberg table. .. tip:: For more details on PyIceberg, see - URI: https://py.iceberg.apache.org/ Examples: .. testcode:: :skipif: True import ray import pandas as pd from ray.data import SaveMode from ray.data.expressions import col # Basic append (default behavior) docs = [{"id": i, "title": f"Doc {i}"} for i in range(4)] ds = ray.data.from_pandas(pd.DataFrame(docs)) ds.write_iceberg( table_identifier="db_name.table_name", catalog_kwargs={"name": "default", "type": "sql"} ) # Schema evolution is automatic - new columns are added automatically enriched_docs = [{"id": i, "title": f"Doc {i}", "category": "new"} for i in range(3)] ds_enriched = ray.data.from_pandas(pd.DataFrame(enriched_docs)) ds_enriched.write_iceberg( table_identifier="db_name.table_name", catalog_kwargs={"name": "default", "type": "sql"} ) # Upsert mode - update existing rows or insert new ones updated_docs = [{"id": 2, "title": "Updated Doc 2"}, {"id": 5, "title": "New Doc 5"}] ds_updates = ray.data.from_pandas(pd.DataFrame(updated_docs)) ds_updates.write_iceberg( table_identifier="db_name.table_name", catalog_kwargs={"name": "default", "type": "sql"}, mode=SaveMode.UPSERT, upsert_kwargs={"join_cols": ["id"]}, ) # Partial overwrite with Ray Data expressions ds.write_iceberg( table_identifier="events.user_activity", catalog_kwargs={"name": "default", "type": "rest"}, mode=SaveMode.OVERWRITE, overwrite_filter=col("date") >= "2024-10-28" ) Args: table_identifier: Fully qualified table identifier (``db_name.table_name``) catalog_kwargs: Optional arguments to pass to PyIceberg's catalog.load_catalog() function (such as name, type, etc.). For the function definition, see `pyiceberg catalog `_. snapshot_properties: Custom properties to write to snapshot when committing to an iceberg table. mode: Write mode using SaveMode enum. Options: * SaveMode.APPEND (default): Add new data to the table without checking for duplicates. * SaveMode.UPSERT: Update existing rows that match on the join condition (``join_cols`` in ``upsert_kwargs``), or insert new rows if they don't exist in the table. * SaveMode.OVERWRITE: Replace all existing data in the table with new data, or replace data matching overwrite_filter if specified. overwrite_filter: Optional filter for OVERWRITE mode to perform partial overwrites. Must be a Ray Data expression from `ray.data.expressions`. Only rows matching this filter are replaced. If None with OVERWRITE mode, replaces all table data. Example: `col("date") >= "2024-01-01"` or `(col("region") == "US") & (col("status") == "active")` upsert_kwargs: Optional arguments for upsert operations. Supported parameters: join_cols (List[str]), case_sensitive (bool), branch (str). Note: Ray Data uses a copy-on-write strategy that always updates all columns for matched keys and inserts all new keys for optimal parallelism. overwrite_kwargs: Optional arguments to pass through to PyIceberg's table.overwrite() method. Supported parameters: case_sensitive (bool), branch (str). See PyIceberg documentation for details. ray_remote_args: kwargs passed to :func:`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. Note: Schema evolution is automatically enabled. New columns in the incoming data are automatically added to the table schema. The schema is extracted automatically from the data being written. )rrrrrrrrN)rr) rrrrrrrrrrrs r write_icebergzDataset.write_icebergZs`J#-) 3-'-     +#      rpng)rrrrrrr file_formatc pt||||||||j|  } || || dS)a Writes the :class:`~ray.data.Dataset` to images. Examples: >>> import ray >>> ds = ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple") >>> ds.write_images("local:///tmp/images", column="image") Time complexity: O(dataset size / parallelism) Args: path: The path to the destination root directory, where the images are written to. column: The column containing the data you want to write to images. file_format: The image file format to write with. For available options, see `Image file formats `_. filesystem: The pyarrow filesystem implementation to write to. These filesystems are specified in the `pyarrow docs `_. Specify this 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. try_create_dir: If ``True``, attempts to create all directories in the destination path. Does nothing if all directories already exist. Defaults to ``True``. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_output_stream `_, which is used when opening the file to write to. filename_provider: A :class:`~ray.data.datasource.FilenameProvider` implementation. Use this parameter to customize what your filenames look like. ray_remote_args: kwargs passed to :func:`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. mode: Determines how to handle existing files. Valid modes are "overwrite", "error", "ignore", "append". Defaults to "append". NOTE: This method isn't atomic. "Overwrite" first deletes all the data before writing to `path`. )rrrrrrrN)rrqr) rrrrrrrrrrrrs r write_imageszDataset.write_imagesshz!   !)3/      +#      r) rrrrarrow_csv_args_fnrrrrrrc |d}t| |\} }t||| | |||||j|  }|||| dS)a3Writes the :class:`~ray.data.Dataset` to CSV files. The number of files is determined by the number of blocks in the dataset. To control the number of number of blocks, call :meth:`~ray.data.Dataset.repartition`. This method is only supported for datasets with records that are convertible to pyarrow tables. By default, the format of the output files is ``{uuid}_{block_idx}.csv``, where ``uuid`` is a unique id for the dataset. To modify this behavior, implement a custom :class:`~ray.data.datasource.FilenameProvider` and pass it in as the ``filename_provider`` argument. Examples: Write the dataset as CSV files to a local directory. >>> import ray >>> ds = ray.data.range(100) >>> ds.write_csv("local:///tmp/data") Write the dataset as CSV files to S3. >>> import ray >>> ds = ray.data.range(100) >>> ds.write_csv("s3://bucket/folder/) # doctest: +SKIP Time complexity: O(dataset size / parallelism) Args: path: The path to the destination root directory, where the CSV files are written to. filesystem: The pyarrow filesystem implementation to write to. These filesystems are specified in the `pyarrow docs `_. Specify this 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. try_create_dir: If ``True``, attempts to create all directories in the destination path if ``True``. Does nothing if all directories already exist. Defaults to ``True``. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_output_stream `_, which is used when opening the file to write to. filename_provider: A :class:`~ray.data.datasource.FilenameProvider` implementation. Use this parameter to customize what your filenames look like. arrow_csv_args_fn: Callable that returns a dictionary of write arguments that are provided to `pyarrow.write.write_csv `_ when writing each block to a file. Overrides any duplicate keys from ``arrow_csv_args``. Use this argument instead of ``arrow_csv_args`` if any of your write arguments cannot be pickled, or if you'd like to lazily resolve the write arguments for each dataset block. min_rows_per_file: [Experimental] The target minimum number of rows to write to each file. If ``None``, Ray Data writes a system-chosen number of rows to each file. If the number of rows per block is larger than the specified value, Ray Data writes the number of rows per block to each file. The specified value is a hint, not a strict limit. Ray Data might write more or fewer rows to each file. ray_remote_args: kwargs passed to :func:`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. num_rows_per_file: [Deprecated] Use min_rows_per_file instead. arrow_csv_args: Options to pass to `pyarrow.write.write_csv `_ when writing each block to a file. mode: Determines how to handle existing files. Valid modes are "overwrite", "error", "ignore", "append". Defaults to "append". NOTE: This method isn't atomic. "Overwrite" first deletes all the data before writing to `path`. NciSrrrrrrz#Dataset.write_csv..srr) rarrow_csv_argsrrrrrrrr)rJrrqr)rrrrrrrrrrrrrrrrs r write_csvzDataset.write_csvsF  $ *  </CT! ! ! A /)0!)3/      +#      r) tf_schemarrrrrrrrrrzschema_pb2.Schemac t| |\} } t||| |||||j|  }|||| dS)aWrite the :class:`~ray.data.Dataset` to TFRecord files. The `TFRecord `_ files contain `tf.train.Example `_ records, with one Example record for each row in the dataset. .. warning:: tf.train.Feature only natively stores ints, floats, and bytes, so this function only supports datasets with these data types, and will error if the dataset contains unsupported types. The number of files is determined by the number of blocks in the dataset. To control the number of number of blocks, call :meth:`~ray.data.Dataset.repartition`. This method is only supported for datasets with records that are convertible to pyarrow tables. By default, the format of the output files is ``{uuid}_{block_idx}.tfrecords``, where ``uuid`` is a unique id for the dataset. To modify this behavior, implement a custom :class:`~ray.data.datasource.FilenameProvider` and pass it in as the ``filename_provider`` argument. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.write_tfrecords("local:///tmp/data/") Time complexity: O(dataset size / parallelism) Args: path: The path to the destination root directory, where tfrecords files are written to. filesystem: The pyarrow filesystem implementation to write to. These filesystems are specified in the `pyarrow docs `_. Specify this 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. try_create_dir: If ``True``, attempts to create all directories in the destination path. Does nothing if all directories already exist. Defaults to ``True``. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_output_stream `_, which is used when opening the file to write to. filename_provider: A :class:`~ray.data.datasource.FilenameProvider` implementation. Use this parameter to customize what your filenames look like. min_rows_per_file: [Experimental] The target minimum number of rows to write to each file. If ``None``, Ray Data writes a system-chosen number of rows to each file. If the number of rows per block is larger than the specified value, Ray Data writes the number of rows per block to each file. The specified value is a hint, not a strict limit. Ray Data might write more or fewer rows to each file. ray_remote_args: kwargs passed to :func:`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. num_rows_per_file: [Deprecated] Use min_rows_per_file instead. mode: Determines how to handle existing files. Valid modes are "overwrite", "error", "ignore", "append". Defaults to "append". NOTE: This method isn't atomic. "Overwrite" first deletes all the data before writing to `path`. r) rrrrrrrrrrN)rJr#rqr)rrrrrrrrrrrrrrrs rwrite_tfrecordszDataset.write_tfrecordssp!=/CT! ! ! A$0!)3/      +#      r) rrrrrrencoderrrrrc t| |\} } t||| |||||j|  }|||| dS)a] Writes the dataset to `WebDataset `_ files. The `TFRecord `_ files will contain `tf.train.Example `_ # noqa: E501 records, with one Example record for each row in the dataset. .. warning:: tf.train.Feature only natively stores ints, floats, and bytes, so this function only supports datasets with these data types, and will error if the dataset contains unsupported types. This is only supported for datasets convertible to Arrow records. To control the number of files, use :meth:`Dataset.repartition`. Unless a custom filename provider is given, the format of the output files is ``{uuid}_{block_idx}.tfrecords``, where ``uuid`` is a unique id for the dataset. Examples: .. testcode:: :skipif: True import ray ds = ray.data.range(100) ds.write_webdataset("s3://bucket/folder/") Time complexity: O(dataset size / parallelism) Args: path: The path to the destination root directory, where tfrecords files are written to. filesystem: The filesystem implementation to write to. try_create_dir: If ``True``, attempts to create all directories in the destination path. Does nothing if all directories already exist. Defaults to ``True``. arrow_open_stream_args: kwargs passed to ``pyarrow.fs.FileSystem.open_output_stream`` filename_provider: A :class:`~ray.data.datasource.FilenameProvider` implementation. Use this parameter to customize what your filenames look like. min_rows_per_file: [Experimental] The target minimum number of rows to write to each file. If ``None``, Ray Data writes a system-chosen number of rows to each file. If the number of rows per block is larger than the specified value, Ray Data writes the number of rows per block to each file. The specified value is a hint, not a strict limit. Ray Data might write more or fewer rows to each file. ray_remote_args: Kwargs passed to :func:`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. num_rows_per_file: [Deprecated] Use min_rows_per_file instead. mode: Determines how to handle existing files. Valid modes are "overwrite", "error", "ignore", "append". Defaults to "append". NOTE: This method isn't atomic. "Overwrite" first deletes all the data before writing to `path`. r)rrrrrrrrrN)rJr$rqr)rrrrrrrrrrrrrrrs rwrite_webdatasetzDataset.write_webdatasetsZ!=/CT! ! ! A& 0!)3/      +#      r) rrrrrrrrrc t| |\} } t||| |||||j|  }|||| dS)a8Writes a column of the :class:`~ray.data.Dataset` to .npy files. This is only supported for columns in the datasets that can be converted to NumPy arrays. The number of files is determined by the number of blocks in the dataset. To control the number of number of blocks, call :meth:`~ray.data.Dataset.repartition`. By default, the format of the output files is ``{uuid}_{block_idx}.npy``, where ``uuid`` is a unique id for the dataset. To modify this behavior, implement a custom :class:`~ray.data.datasource.FilenameProvider` and pass it in as the ``filename_provider`` argument. Examples: >>> import ray >>> ds = ray.data.range(100) >>> ds.write_numpy("local:///tmp/data/", column="id") Time complexity: O(dataset size / parallelism) Args: path: The path to the destination root directory, where the npy files are written to. column: The name of the column that contains the data to be written. filesystem: The pyarrow filesystem implementation to write to. These filesystems are specified in the `pyarrow docs `_. Specify this 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. try_create_dir: If ``True``, attempts to create all directories in destination path. Does nothing if all directories already exist. Defaults to ``True``. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_output_stream `_, which is used when opening the file to write to. filename_provider: A :class:`~ray.data.datasource.FilenameProvider` implementation. Use this parameter to customize what your filenames look like. min_rows_per_file: [Experimental] The target minimum number of rows to write to each file. If ``None``, Ray Data writes a system-chosen number of rows to each file. If the number of rows per block is larger than the specified value, Ray Data writes the number of rows per block to each file. The specified value is a hint, not a strict limit. Ray Data might write more or fewer rows to each file. ray_remote_args: kwargs passed to :func:`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. num_rows_per_file: [Deprecated] Use min_rows_per_file instead. mode: Determines how to handle existing files. Valid modes are "overwrite", "error", "ignore", "append". Defaults to "append". NOTE: This method isn't atomic. "Overwrite" first deletes all the data before writing to `path`. r)rrrrrrrrN)rJr rqr)rrrrrrrrrrrrrrrs r write_numpyzDataset.write_numpyhs`!=/CT! ! ! A!  0!)3/      +#      rsqlconnection_factorycXt||}||||dS)aWrite to a database that provides a `Python DB API2-compliant `_ connector. .. note:: This method writes data in parallel using the DB API2 ``executemany`` method. To learn more about this method, see `PEP 249 `_. Examples: .. testcode:: import sqlite3 import ray connection = sqlite3.connect("example.db") connection.cursor().execute("CREATE TABLE movie(title, year, score)") dataset = ray.data.from_items([ {"title": "Monty Python and the Holy Grail", "year": 1975, "score": 8.2}, {"title": "And Now for Something Completely Different", "year": 1971, "score": 7.5} ]) dataset.write_sql( "INSERT INTO movie VALUES(?, ?, ?)", lambda: sqlite3.connect("example.db") ) result = connection.cursor().execute("SELECT * FROM movie ORDER BY year") print(result.fetchall()) .. testoutput:: [('And Now for Something Completely Different', 1971, 7.5), ('Monty Python and the Holy Grail', 1975, 8.2)] .. testcode:: :hide: import os os.remove("example.db") Arguments: sql: An ``INSERT INTO`` statement that specifies the table to write to. The number of parameters must match the number of columns in the table. connection_factory: A function that takes no arguments and returns a Python DB API2 `Connection object `_. ray_remote_args: Keyword arguments passed to :func:`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. )rrrN)r"r)rrrrrrs r write_sqlzDataset.write_sqlsHz3;MNNN  +#      rrtableconnection_parametersc$ ddl  fd}|j}dd|D}ddgt |z}d|d|d |d } || ||| dS) aSWrite this ``Dataset`` to a Snowflake table. Examples: .. testcode:: :skipif: True import ray connection_parameters = dict( user=..., account="ABCDEFG-ABC12345", password=..., database="SNOWFLAKE_SAMPLE_DATA", schema="TPCDS_SF100TCL" ) ds = ray.data.read_parquet("s3://anonymous@ray-example-data/iris.parquet") ds.write_snowflake("MY_DATABASE.MY_SCHEMA.IRIS", connection_parameters) Args: table: The name of the table to write to. connection_parameters: Keyword arguments to pass to ``snowflake.connector.connect``. To view supported parameters, read https://docs.snowflake.com/developer-guide/python-connector/python-connector-api#functions. ray_remote_args: Keyword arguments passed to :func:`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. rNc(jjdiS)Nr) connectorconnect)r snowflakesrsnowflake_connection_factoryz=Dataset.write_snowflake..snowflake_connection_factory;s .9&.GG1FGG Grz, c3"K|] }d|dV dS)"Nrr s rr z*Dataset.write_snowflake..Bs*CCs C CCCCCCrz%sz INSERT INTO z (z ) VALUES (r<)rrr)snowflake.connectorrrrFrr) rrrrrrr columns_str placeholdersrrs ` @rwrite_snowflakezDataset.write_snowflakesP #""" H H H H H H{{}}* iiCClCCCCC yy$#l*;*;!;<< LULLkLL\LLL  ;+#      ruridatabase collectioncZt|||}||||dS)a Writes the :class:`~ray.data.Dataset` to a MongoDB database. This method is only supported for datasets convertible to pyarrow tables. The number of parallel writes is determined by the number of blocks in the dataset. To control the number of number of blocks, call :meth:`~ray.data.Dataset.repartition`. .. warning:: This method supports only a subset of the PyArrow's types, due to the limitation of pymongoarrow which is used underneath. Writing unsupported types fails on type checking. See all the supported types at: https://mongo-arrow.readthedocs.io/en/stable/api/types.html. .. note:: The records are inserted into MongoDB as new documents. If a record has the _id field, this _id must be non-existent in MongoDB, otherwise the write is rejected and fail (hence preexisting documents are protected from being mutated). It's fine to not have _id field in record and MongoDB will auto generate one at insertion. Examples: .. testcode:: :skipif: True import ray ds = ray.data.range(100) ds.write_mongo( uri="mongodb://username:password@mongodb0.example.com:27017/?authSource=admin", database="my_db", collection="my_collection" ) Args: uri: The URI to the destination MongoDB where the dataset is written to. For the URI format, see details in the `MongoDB docs `_. database: The name of the database. This database must exist otherwise a ValueError is raised. collection: The name of the collection in the database. This collection must exist otherwise a ValueError is raised. ray_remote_args: kwargs passed to :func:`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. Raises: ValueError: if ``database`` doesn't exist. ValueError: if ``collection`` doesn't exist. )rrrrN)rr)rrrrrrrs r write_mongozDataset.write_mongoLsT@!!     +#      r project_idr max_retry_cntoverwrite_tablec|i}|dddkrtjdnd|d<t||||}||||dS)aWrite the dataset to a BigQuery dataset table. To control the number of parallel write tasks, use ``.repartition()`` before calling this method. Examples: .. testcode:: :skipif: True import ray import pandas as pd docs = [{"title": "BigQuery Datasource test"} for key in range(4)] ds = ray.data.from_pandas(pd.DataFrame(docs)) ds.write_bigquery( project_id="my_project_id", dataset="my_dataset_table", overwrite_table=True ) Args: project_id: The name of the associated Google Cloud Project that hosts the dataset to read. For more information, see details in `Creating and managing projects `_. dataset: The name of the dataset in the format of ``dataset_id.table_id``. The dataset is created if it doesn't already exist. max_retry_cnt: The maximum number of retries that an individual block write is retried due to BigQuery rate limiting errors. This isn't related to Ray fault tolerance retries. The default number of retries is 10. overwrite_table: Whether the write will overwrite the table if it already exists. The default behavior is to overwrite the table. ``overwrite_table=False`` will append to the table if it exists. ray_remote_args: Kwargs passed to :func:`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. N max_retriesrzVThe max_retries of a BigQuery Write Task should be set to 0 to avoid duplicate writes.)rrrrr)rrrrr)rrrrrrrrs rwrite_bigqueryzDataset.write_bigquerysb  " O   }a 0 0A 5 5 M.     ./OM *#!'+      +#      r)rrclient_settings client_kwargstable_settingsmax_insert_block_rowsrrdsnzpyarrow.Schemar r r r c dt||||||||} || | | dS)a%Write the dataset to a ClickHouse dataset table. To control the number of parallel write tasks, use ``.repartition()`` before calling this method. Examples: .. testcode:: :skipif: True import ray import pyarrow as pa import pandas as pd docs = [{"title": "ClickHouse Datasink test"} for key in range(4)] ds = ray.data.from_pandas(pd.DataFrame(docs)) user_schema = pa.schema( [ ("id", pa.int64()), ("title", pa.string()), ] ) ds.write_clickhouse( table="default.my_table", dsn="clickhouse+http://user:pass@localhost:8123/default", mode=ray.data.SinkMode.OVERWRITE, schema=user_schema, table_settings=ray.data.ClickHouseTableSettings( engine="ReplacingMergeTree()", order_by="id", ), ) Args: table: Fully qualified table identifier (e.g., "default.my_table"). The table is created if it doesn't already exist. dsn: A string in DSN (Data Source Name) HTTP format (e.g., "clickhouse+http://username:password@host:8123/default"). For more information, see `ClickHouse Connection String doc `_. mode: One of SinkMode.CREATE, SinkMode.APPEND, or SinkMode.OVERWRITE: * SinkMode.CREATE: Create a new table; fail if it already exists. If the table does not exist, you must provide a schema (either via the `schema` argument or as part of the dataset's first block). * SinkMode.APPEND: If the table exists, append data to it; if not, create the table using the provided or inferred schema. If the table does not exist, you must supply a schema. * SinkMode.OVERWRITE: Drop any existing table of this name, then create a new table and write data to it. You **must** provide a schema in this case, as the table is being re-created. schema: Optional :class:`pyarrow.Schema` specifying column definitions. This is mandatory if you are creating a new table (i.e., table doesn't exist in CREATE or APPEND mode) or overwriting an existing table (OVERWRITE). When appending to an existing table, a schema is optional, though you can provide one to enforce column types or cast data as needed. If omitted (and the table already exists), the existing table definition will be used. If omitted and the table must be created, the schema is inferred from the first block in the dataset. client_settings: Optional ClickHouse server settings to be used with the session/every request. For more information, see `ClickHouse Client Settings doc `_. client_kwargs: Optional keyword arguments to pass to the ClickHouse client. For more information, see `ClickHouse Core Settings doc `_. table_settings: An optional :class:`ClickHouseTableSettings` dataclass that specifies additional table creation instructions, including: * engine (default: `"MergeTree()"`): Specifies the engine for the `CREATE TABLE` statement. * order_by: Sets the `ORDER BY` clause in the `CREATE TABLE` statement, iff not provided. When overwriting an existing table, its previous `ORDER BY` (if any) is reused. Otherwise, a "best" column is selected automatically (favoring a timestamp column, then a non-string column, and lastly the first column). * partition_by: If present, adds a `PARTITION BY ` clause to the `CREATE TABLE` statement. * primary_key: If present, adds a `PRIMARY KEY ()` clause. * settings: Appends a `SETTINGS ` clause to the `CREATE TABLE` statement, allowing custom ClickHouse settings. max_insert_block_rows: If you have extremely large blocks, specifying a limit here will chunk the insert into multiple smaller insert calls. Defaults to None (no chunking). ray_remote_args: Kwargs passed to :func:`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. )rrrrr r r r rN)rr) rrrrrr r r r rrrs rwrite_clickhousezDataset.write_clickhousesch&+')"7      +#      rrrii)rrrrdata_storage_versionstorage_optionsrr)rrr overwriterrc bt|||||||} || || dS)a\Write the dataset to a Lance dataset. Examples: .. testcode:: import ray import pandas as pd docs = [{"title": "Lance data sink test"} for key in range(4)] ds = ray.data.from_pandas(pd.DataFrame(docs)) ds.write_lance("/tmp/data/") Args: path: The path to the destination Lance dataset. schema: The schema of the dataset. If not provided, it is inferred from the data. mode: The write mode. Can be "create", "append", or "overwrite". min_rows_per_file: The minimum number of rows per file. max_rows_per_file: The maximum number of rows per file. data_storage_version: The version of the data storage format to use. Newer versions are more efficient but require newer versions of lance to read. The default is "legacy" which will use the legacy v1 version. See the user guide for more details. storage_options: The storage options for the writer. Default is None. )rrrrrrrN)rr) rrrrrrrrrrrs r write_lancezDataset.write_lancefs`J! //!5+     +#      r)rrcf|i}|jsntjjjrt dt tjd|d<|j }t|j j |||}t||j} t!|t"r@||jr%t(d|jddSt/|||_|j\}}g} |D]=} tj| j} | t=| >t?j | } t(d |!| j"tG| j$|%| dS#tL$r} |'| d} ~ wwxYw) a^Writes the dataset to a custom :class:`~ray.data.Datasink`. Time complexity: O(dataset size / parallelism) Args: datasink: The :class:`~ray.data.Datasink` to write to. ray_remote_args: Kwargs passed to :func:`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. NzUIf you're using Ray Client, Ray Data won't schedule write tasks on the driver's node.F)softscheduling_strategyrzIgnoring write because z already existsz3Data sink %s finished. %d rows and %s data written.)(supports_distributed_writesrrclient is_connectedrrmget_runtime_context get_node_idrrr=rrr,rrreon_write_start _skip_writerrrrrr_execute_to_iteratorrrrrdrccombineget_namer r(ron_write_complete Exceptionon_write_failed)rrrrrwrite_opriter_r write_resultsrrcombined_write_resultes rrzDataset.write_datasinksM(  " O3 x"//11  ,6T'))5577666O1 2 z     " +#     #8T\:: $  (M22 '')))'KKP(-PPPF$T<88DDFFDN>>>@@LE5M F Fgf/00$$%?%D%DEEEE$/$7$G ! KKE!!##%.3>??      & &'< = = = = =     $ $Q ' ' '  s AH C.H H0H++H0zGCalling any of the consumption methods on the returned ``DataIterator``zReturns:)delegaterc t|S)zReturn a :class:`~ray.data.DataIterator` over this dataset. Don't call this method directly. Use it internally. Returns: A :class:`~ray.data.DataIterator` over this dataset. r)rs riteratorzDataset.iterators %%%rcN|S)akReturn an iterable over the rows in this dataset. Examples: >>> import ray >>> for row in ray.data.range(3).iter_rows(): ... print(row) {'id': 0} {'id': 1} {'id': 2} Time complexity: O(1) Returns: An iterable over the rows in this dataset. )r-rrs rrzDataset.iter_rowss$}}((***rrrr drop_lastlocal_shuffle_buffer_sizelocal_shuffle_seed _collate_fnrr1r2r3r4c |t|}||||||||S)aReturn an iterable over batches of data. This method is useful for model training. Examples: .. testcode:: import ray ds = ray.data.read_images("example://image-datasets/simple") for batch in ds.iter_batches(batch_size=2, batch_format="numpy"): print(batch) .. testoutput:: :options: +MOCK {'image': array([[[[...]]]], dtype=uint8)} ... {'image': array([[[[...]]]], dtype=uint8)} Time complexity: O(1) Args: prefetch_batches: The number of batches to fetch ahead of the current batch to fetch. If set to greater than 0, a separate threadpool is used to fetch the objects to the local node and format the batches. Defaults to 1. batch_size: The number of rows in each batch, or ``None`` to use entire blocks as batches (blocks may contain different numbers of rows). The final batch may include fewer than ``batch_size`` rows if ``drop_last`` is ``False``. Defaults to 256. batch_format: If ``"default"`` or ``"numpy"``, batches are ``Dict[str, numpy.ndarray]``. If ``"pandas"``, batches are ``pandas.DataFrame``. drop_last: Whether to drop the last batch if it's incomplete. local_shuffle_buffer_size: If not ``None``, the data is randomly shuffled using a local in-memory shuffle buffer, and this value serves as the minimum number of rows that must be in the local in-memory shuffle buffer in order to yield a batch. When there are no more rows to add to the buffer, the remaining rows in the buffer are drained. local_shuffle_seed: The seed to use for the local random shuffle. Returns: An iterable over batches of data. r0)r]r- _iter_batches)rrrrr1r2r3r4s rrzDataset.iter_batchessLx+<88 }},,-!%&?1#-   rautorrdtypesdevice collate_fnr1r2r3r9z torch.dtyper:r;c `|||||||||S)aReturn an iterable over batches of data represented as Torch tensors. This iterable yields batches of type ``Dict[str, torch.Tensor]``. For more flexibility, call :meth:`~Dataset.iter_batches` and manually convert your data to Torch tensors. Examples: >>> import ray >>> for batch in ray.data.range( ... 12, ... ).iter_torch_batches(batch_size=4): ... print(batch) {'id': tensor([0, 1, 2, 3])} {'id': tensor([4, 5, 6, 7])} {'id': tensor([ 8, 9, 10, 11])} Use the ``collate_fn`` to customize how the tensor batch is created. >>> from typing import Any, Dict >>> import torch >>> import numpy as np >>> import ray >>> def collate_fn(batch: Dict[str, np.ndarray]) -> Any: ... return torch.stack( ... [torch.as_tensor(array) for array in batch.values()], ... axis=1 ... ) >>> dataset = ray.data.from_items([ ... {"col_1": 1, "col_2": 2}, ... {"col_1": 3, "col_2": 4}]) >>> for batch in dataset.iter_torch_batches(collate_fn=collate_fn): ... print(batch) tensor([[1, 2], [3, 4]]) Time complexity: O(1) Args: prefetch_batches: The number of batches to fetch ahead of the current batch to fetch. If set to greater than 0, a separate threadpool is used to fetch the objects to the local node, format the batches, and apply the ``collate_fn``. Defaults to 1. batch_size: The number of rows in each batch, or ``None`` to use entire blocks as batches (blocks may contain different number of rows). The final batch may include fewer than ``batch_size`` rows if ``drop_last`` is ``False``. Defaults to 256. dtypes: The Torch dtype(s) for the created tensor(s); if ``None``, the dtype is inferred from the tensor data. You can't use this parameter with ``collate_fn``. device: The device on which the tensor should be placed. Defaults to "auto" which moves the tensors to the appropriate device when the Dataset is passed to Ray Train and ``collate_fn`` is not provided. Otherwise, defaults to CPU. You can't use this parameter with ``collate_fn``. collate_fn: A function to convert a Numpy batch to a PyTorch tensor batch. When this parameter is specified, the user should manually handle the host to device data transfer outside of collate_fn. This is useful for further processing the data after it has been batched. Potential use cases include collating along a dimension other than the first, padding sequences of various lengths, or generally handling batches of different length tensors. If not provided, the default collate function is used which simply converts the batch of numpy arrays to a batch of PyTorch tensors. This API is still experimental and is subject to change. You can't use this parameter in conjunction with ``dtypes`` or ``device``. drop_last: Whether to drop the last batch if it's incomplete. local_shuffle_buffer_size: If not ``None``, the data is randomly shuffled using a local in-memory shuffle buffer, and this value serves as the minimum number of rows that must be in the local in-memory shuffle buffer in order to yield a batch. When there are no more rows to add to the buffer, the remaining rows in the buffer are drained. ``batch_size`` must also be specified when using local shuffling. local_shuffle_seed: The seed to use for the local random shuffle. Returns: An iterable over Torch Tensor batches. .. seealso:: :meth:`Dataset.iter_batches` Call this method to manually convert your data to Torch tensors. r8)r-iter_torch_batches) rrrr9r:r;r1r2r3s rr=zDataset.iter_torch_batchesXsB@}}11-!!&?12   rrrr9r1r2r3ztf.dtypes.DTypectjdt|||||||S)a1 Return an iterable over batches of data represented as TensorFlow tensors. This iterable yields batches of type ``Dict[str, tf.Tensor]``. For more flexibility, call :meth:`~Dataset.iter_batches` and manually convert your data to TensorFlow tensors. .. tip:: If you don't need the additional flexibility provided by this method, consider using :meth:`~ray.data.Dataset.to_tf` instead. It's easier to use. Examples: .. testcode:: import ray ds = ray.data.read_csv("s3://anonymous@air-example-data/iris.csv") tf_dataset = ds.to_tf( feature_columns="sepal length (cm)", label_columns="target", batch_size=2 ) for features, labels in tf_dataset: print(features, labels) .. testoutput:: tf.Tensor([5.1 4.9], shape=(2,), dtype=float64) tf.Tensor([0 0], shape=(2,), dtype=int64) ... tf.Tensor([6.2 5.9], shape=(2,), dtype=float64) tf.Tensor([2 2], shape=(2,), dtype=int64) Time complexity: O(1) Args: prefetch_batches: The number of batches to fetch ahead of the current batch to fetch. If set to greater than 0, a separate threadpool is used to fetch the objects to the local node, format the batches, and apply the ``collate_fn``. Defaults to 1. batch_size: The number of rows in each batch, or ``None`` to use entire blocks as batches (blocks may contain different numbers of rows). The final batch may include fewer than ``batch_size`` rows if ``drop_last`` is ``False``. Defaults to 256. dtypes: The TensorFlow dtype(s) for the created tensor(s); if ``None``, the dtype is inferred from the tensor data. drop_last: Whether to drop the last batch if it's incomplete. local_shuffle_buffer_size: If not ``None``, the data is randomly shuffled using a local in-memory shuffle buffer, and this value serves as the minimum number of rows that must be in the local in-memory shuffle buffer in order to yield a batch. When there are no more rows to add to the buffer, the remaining rows in the buffer are drained. ``batch_size`` must also be specified when using local shuffling. local_shuffle_seed: The seed to use for the local random shuffle. Returns: An iterable over TensorFlow Tensor batches. .. seealso:: :meth:`Dataset.iter_batches` Call this method to manually convert your data to TensorFlow tensors. zX`iter_tf_batches` is deprecated and will be removed after May 2025. Use `to_tf` instead.r>)rrrr-iter_tf_batches)rrrr9r1r2r3s rr@zDataset.iter_tf_batchessZT       }}..-!&?1 /   r) additional_columnsrrr1r2r3feature_type_speclabel_type_specadditional_type_specfeature_columns label_columnsrArBrzrCrDztf.data.Datasetc f|||||||||| | |  S)aZReturn a `TensorFlow Dataset `_ over this :class:`~ray.data.Dataset`. .. warning:: If your :class:`~ray.data.Dataset` contains ragged tensors, this method errors. To prevent errors, :ref:`resize your tensors `. Examples: >>> import ray >>> ds = ray.data.read_csv("s3://anonymous@air-example-data/iris.csv") >>> ds Dataset(num_rows=?, schema=...) If your model accepts a single tensor as input, specify a single feature column. >>> ds.to_tf(feature_columns="sepal length (cm)", label_columns="target") <_OptionsDataset element_spec=(TensorSpec(shape=(None,), dtype=tf.float64, name='sepal length (cm)'), TensorSpec(shape=(None,), dtype=tf.int64, name='target'))> If your model accepts a dictionary as input, specify a list of feature columns. >>> ds.to_tf(["sepal length (cm)", "sepal width (cm)"], "target") <_OptionsDataset element_spec=({'sepal length (cm)': TensorSpec(shape=(None,), dtype=tf.float64, name='sepal length (cm)'), 'sepal width (cm)': TensorSpec(shape=(None,), dtype=tf.float64, name='sepal width (cm)')}, TensorSpec(shape=(None,), dtype=tf.int64, name='target'))> If your dataset contains multiple features but your model accepts a single tensor as input, combine features with :class:`~ray.data.preprocessors.Concatenator`. >>> from ray.data.preprocessors import Concatenator >>> columns_to_concat = ["sepal length (cm)", "sepal width (cm)", "petal length (cm)", "petal width (cm)"] >>> preprocessor = Concatenator(columns=columns_to_concat, output_column_name="features") >>> ds = preprocessor.transform(ds) >>> ds Concatenator +- Dataset(num_rows=?, schema=...) >>> ds.to_tf("features", "target") <_OptionsDataset element_spec=(TensorSpec(shape=(None, 4), dtype=tf.float64, name='features'), TensorSpec(shape=(None,), dtype=tf.int64, name='target'))> If your model accepts different types, shapes, or names of tensors as input, specify the type spec. If type specs are not specified, they are automatically inferred from the schema of the dataset. >>> import tensorflow as tf >>> ds.to_tf( ... feature_columns="features", ... label_columns="target", ... feature_type_spec=tf.TensorSpec(shape=(None, 4), dtype=tf.float32, name="features"), ... label_type_spec=tf.TensorSpec(shape=(None,), dtype=tf.float32, name="label") ... ) <_OptionsDataset element_spec=(TensorSpec(shape=(None, 4), dtype=tf.float32, name='features'), TensorSpec(shape=(None,), dtype=tf.float32, name='label'))> If your model accepts additional metadata aside from features and label, specify a single additional column or a list of additional columns. A common use case is to include sample weights in the data samples and train a ``tf.keras.Model`` with ``tf.keras.Model.fit``. >>> import pandas as pd >>> ds = ds.add_column("sample weights", lambda df: pd.Series([1] * len(df))) >>> ds.to_tf(feature_columns="features", label_columns="target", additional_columns="sample weights") <_OptionsDataset element_spec=(TensorSpec(shape=(None, 4), dtype=tf.float64, name='features'), TensorSpec(shape=(None,), dtype=tf.int64, name='target'), TensorSpec(shape=(None,), dtype=tf.int64, name='sample weights'))> If your model accepts different types, shapes, or names for the additional metadata, specify the type spec of the additional column. >>> ds.to_tf( ... feature_columns="features", ... label_columns="target", ... additional_columns="sample weights", ... additional_type_spec=tf.TensorSpec(shape=(None,), dtype=tf.float32, name="weight") ... ) <_OptionsDataset element_spec=(TensorSpec(shape=(None, 4), dtype=tf.float64, name='features'), TensorSpec(shape=(None,), dtype=tf.int64, name='target'), TensorSpec(shape=(None,), dtype=tf.float32, name='weight'))> Args: feature_columns: Columns that correspond to model inputs. If this is a string, the input data is a tensor. If this is a list, the input data is a ``dict`` that maps column names to their tensor representation. label_columns: Columns that correspond to model targets. If this is a string, the target data is a tensor. If this is a list, the target data is a ``dict`` that maps column names to their tensor representation. additional_columns: Columns that correspond to sample weights or other metadata. If this is a string, the weight data is a tensor. If this is a list, the weight data is a ``dict`` that maps column names to their tensor representation. prefetch_batches: The number of batches to fetch ahead of the current batch to fetch. If set to greater than 0, a separate threadpool is used to fetch the objects to the local node, format the batches, and apply the collate_fn. Defaults to 1. batch_size: Record batch size. Defaults to 1. drop_last: Set to True to drop the last incomplete batch, if the dataset size is not divisible by the batch size. If False and the size of the stream is not divisible by the batch size, then the last batch is smaller. Defaults to False. local_shuffle_buffer_size: If non-None, the data is randomly shuffled using a local in-memory shuffle buffer, and this value will serve as the minimum number of rows that must be in the local in-memory shuffle buffer in order to yield a batch. When there are no more rows to add to the buffer, the remaining rows in the buffer is drained. This buffer size must be greater than or equal to ``batch_size``, and therefore ``batch_size`` must also be specified when using local shuffling. local_shuffle_seed: The seed to use for the local random shuffle. feature_type_spec: The `tf.TypeSpec` of `feature_columns`. If there is only one column, specify a `tf.TypeSpec`. If there are multiple columns, specify a ``dict`` that maps column names to their `tf.TypeSpec`. Default is `None` to automatically infer the type of each column. label_type_spec: The `tf.TypeSpec` of `label_columns`. If there is only one column, specify a `tf.TypeSpec`. If there are multiple columns, specify a ``dict`` that maps column names to their `tf.TypeSpec`. Default is `None` to automatically infer the type of each column. additional_type_spec: The `tf.TypeSpec` of `additional_columns`. If there is only one column, specify a `tf.TypeSpec`. If there are multiple columns, specify a ``dict`` that maps column names to their `tf.TypeSpec`. Default is `None` to automatically infer the type of each column. Returns: A `TensorFlow Dataset`_ that yields inputs and targets. .. seealso:: :meth:`~ray.data.Dataset.iter_tf_batches` Call this method if you need more flexibility. ) rErFrArr1rr2r3rBrCrD)r-to_tf) rrErFrArrr1r2r3rBrCrDs rrHz Dataset.to_tfsKL}}$$+'1-!&?1/+!5%   rdaft.DataFramec*ddl}|j|S)aConvert this :class:`~ray.data.Dataset` into a `Daft DataFrame `_. This will convert all the data inside the Ray Dataset into a Daft DataFrame in a zero-copy way (using Arrow as the intermediate data format). Time complexity: O(dataset size / parallelism) Returns: A `Daft DataFrame`_ created from this dataset. rN)daftfrom_ray_dataset)rrKs rto_daftzDataset.to_dafts   $t$T***rmetapandas.DataFramez pandas.Series verify_metazdask.dataframe.DataFramecddl}ddlm}ddl ddl}n#t $rd}YnwxYwddlm}ddlm ddl m }|j ||jdtt djffd }| dd lm|d } t+| |r;fd t-| j| jD}n|t+| |jrt5t7fd| jDr;fdt-| j| jD}n&| }g} |D]*} | jD] } | || !+|!| ||} | S)aConvert this :class:`~ray.data.Dataset` into a `Dask DataFrame `_. This is only supported for datasets convertible to Arrow records. Note that this function will set the Dask scheduler to Dask-on-Ray globally, via the config. Time complexity: O(dataset size / parallelism) Args: meta: An empty `pandas DataFrame`_ or `Series`_ that matches the dtypes and column names of the stream. This metadata is necessary for many algorithms in dask dataframe to work. For ease of use, some alternative inputs are also available. Instead of a DataFrame, a dict of ``{name: dtype}`` or iterable of ``(name, dtype)`` can be provided (note that the order of the names should match the order of the columns). Instead of a series, a tuple of ``(name, dtype)`` can be used. By default, this is inferred from the underlying Dataset schema, with this argument supplying an optional override. verify_meta: If True, Dask will check that the partitions have consistent metadata. Defaults to True. Returns: A `Dask DataFrame`_ created from this dataset. .. _pandas DataFrame: https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html .. _Series: https://pandas.pydata.org/docs/reference/api/pandas.Series.html rNr?)ClientObjectRef) ray_dask_get) schedulerrrcxt|tjfrtdt |S)NzzDataset.to_dask() must be used with Dask-on-Ray, please set the Dask scheduler to ray_dask_get (located in ray.util.dask).)rrrir _block_to_df)rrSs r block_to_dfz$Dataset.to_dask..block_to_dfsC)cm_%EFF  &  ** *r) TensorDtypeTrHc |i|]8\}}|t|s|n tj9S)dtype)rrrobject_)r rxr\rYrs rrz#Dataset.to_dask.. sa   'CRYY(2%'E'E!0%'Z '   rc38K|]}t|VdSr)r)r type_arrow_tensor_ext_typess rr z"Dataset.to_dask..s?BGJu&<==rc i|]J\}}|t|s|n tjKSr[)rrto_pandas_dtyperr])r rxr\r`rs rrz#Dataset.to_dask..so   !+U ,6e=S+T+T%4E$9$9$;$;$;)+ "+""   r)rNrP)"daskdask.dataframe dataframerrr$ray.data._internal.pandas_blockr?ray.util.client.commonrS ray.util.daskrTconfigrdelayedrirVrray.data.extensionsrYrrr-rtypesrrr empty_table to_pandasiter_internal_ref_bundlesrr from_delayed)rrNrPrcddrr?rTrXrdfs ref_bundlerddfrSrYr`rs @@@@rto_daskzDataset.to_dasksV  ######     BBB  FEEEEE::::::...... ,///  +9U#3 +  + + + + +  + < 7 7 7 7 7 7[[$[77F&"344 <||     +.flFL*I*I     Jvry$A$A)U)W)W&KQ<<<<     /2&, .M.M     DD"--//99;;D88:: 3 3J'2 3 3  ;;y112222 3oo #    s  ((mars.dataframe.DataFramec\ddl}ddl}ddlm}ddlm}ddlm}|}| }t|tr|j }t||r"| |j|j}nSt||j r,|j}nt'd|||d} ||jd } || } | | | | S) a&Convert this :class:`~ray.data.Dataset` into a `Mars DataFrame `_. Time complexity: O(dataset size / parallelism) Returns: A `Mars DataFrame`_ created from this dataset. rN)DataFrameReadRayDataset) parse_indexrR)rzUnsupported format of schema rT) store_data)refs) index_value columns_valuer9)rr)mars.dataframe.datasource.read_raydatasetrxmars.dataframe.utilsryrfr?to_pandas_refsrrrrorrlrrmrnr9r RangeIndexr) rrrrxryr?r{rr9r|r}r2s rto_marszDataset.to_mars8sb UUUUUU444444EEEEEE""$$ ff % % ('F f/ 0 0 PYYv|6`_. This works by first converting this dataset into a distributed set of Pandas DataFrames (using :meth:`Dataset.to_pandas_refs`). See caveats there. Then the individual DataFrames are used to create the Modin DataFrame using ``modin.distributed.dataframe.pandas.partitions.from_partitions()``. This is only supported for datasets convertible to Arrow records. This function induces a copy of the data. For zero-copy access to the underlying data, consider using :meth:`.to_arrow_refs` or :meth:`.iter_internal_ref_bundles`. Time complexity: O(dataset size / parallelism) Returns: A `Modin DataFrame`_ created from this dataset. r)from_partitions)axis)-modin.distributed.dataframe.pandas.partitionsrr)rrpd_objss rto_modinzDataset.to_modinZs=. RQQQQQ%%''wQ////rsparkpyspark.sql.SparkSessionpyspark.sql.DataFramecddl}|}t|tr|j}|}t |}|j|||S)a'Convert this :class:`~ray.data.Dataset` into a `Spark DataFrame `_. Time complexity: O(dataset size / parallelism) Args: spark: A `SparkSession`_, which must be created by RayDP (Spark-on-Ray). Returns: A `Spark DataFrame`_ created from this dataset. .. _SparkSession: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.SparkSession.html rN) raydprrrroror'rray_dataset_to_spark_dataframe)rrrrrrs rto_sparkzDataset.to_sparkvsl   ff % % ('F4466 =kJJ {99%TTTrc Z|3|}||krtd|d|d|dt}|ddD]}|||}t j|S)aConvert this :class:`~ray.data.Dataset` to a single pandas DataFrame. This method errors if the number of rows exceeds the provided ``limit``. To truncate the dataset beforehand, call :meth:`.limit`. Examples: >>> import ray >>> ds = ray.data.from_items([{"a": i} for i in range(3)]) >>> ds.to_pandas() a 0 0 1 1 2 2 Time complexity: O(dataset size) Args: limit: The maximum number of rows to return. An error is raised if the dataset has more rows than this limit. Defaults to ``None``, which means no limit. Returns: A pandas DataFrame created from this dataset, containing a limited number of rows. Raises: ValueError: if the number of rows in the :class:`~ray.data.Dataset` exceeds ``limit``. Nz-the dataset has more than the given limit of z rows: z(. If you are sure that a DataFrame with zO rows will fit in local memory, set ds.to_pandas(limit=None) to disable limits.r)rr) rrr>r add_blockbuildrW for_blockrn)rrrbuilderrblocks rrnzDataset.to_pandass@  JJLLEu}} BEBB"BBBBB%&&&&H&NN % %E   e $ $ $ $  &u--77999rctt}g}|D]4}|jD]*}|||+5|S)aConverts this :class:`~ray.data.Dataset` into a distributed set of Pandas dataframes. One DataFrame is created for each block in this Dataset. This function induces a copy of the data. For zero-copy access to the underlying data, consider using :meth:`Dataset.to_arrow_refs` or :meth:`Dataset.iter_internal_ref_bundles`. Examples: >>> import ray >>> ds = ray.data.range(10, override_num_blocks=2) >>> refs = ds.to_pandas_refs() >>> len(refs) 2 Time complexity: O(dataset size / parallelism) Returns: A list of remote pandas DataFrames created from this dataset. )rBrWrorrremote)rrX pandas_refsrrs rrzDataset.to_pandas_refssx2'|44  4466 B BF#. B B "";#5#5i#@#@AAAA Brrctt}g}|D]6}|jD],}||||-7|S)aConverts this :class:`~ray.data.Dataset` into a distributed set of NumPy ndarrays or dictionary of NumPy ndarrays. This is only supported for datasets convertible to NumPy ndarrays. This function induces a copy of the data. For zero-copy access to the underlying data, consider using :meth:`Dataset.to_arrow_refs` or :meth:`Dataset.iter_internal_ref_bundles`. Examples: >>> import ray >>> ds = ray.data.range(10, override_num_blocks=2) >>> refs = ds.to_numpy_refs() >>> len(refs) 2 Time complexity: O(dataset size / parallelism) Args: column: The name of the column to convert to numpy. If ``None``, all columns are used. If multiple columns are specified, each returned future represents a dict of ndarrays. Defaults to None. Returns: A list of remote NumPy ndarrays created from this dataset. r)rB_block_to_ndarrayrorrr)rrblock_to_ndarray numpy_refsrrs r to_numpy_refszDataset.to_numpy_refss:,,=>> 4466 U UF#. U U !!"2"9"9)F"9"S"STTTT Urz pyarrow.Tablec4ddl}|j}t|g}|d}t |t r|j}t ||jr|Sttfd|DS)aConvert this :class:`~ray.data.Dataset` into a distributed set of PyArrow tables. One PyArrow table is created for each block in this Dataset. This method is only supported for datasets convertible to PyArrow tables. This function is zero-copy if the existing data is already in PyArrow format. Otherwise, the data is converted to PyArrow format. Examples: >>> import ray >>> ds = ray.data.range(10, override_num_blocks=2) >>> refs = ds.to_arrow_refs() >>> len(refs) 2 Time complexity: O(1) unless conversion is required. Returns: A list of remote PyArrow tables created from this dataset. rNTrHc:g|]}|Sr)r)r rblock_to_arrows rrz)Dataset.to_arrow_refs..0s'EEE%%e,,EEEr) rrrr'rrrrorB_block_to_arrow)rrrsrrrs @r to_arrow_refszDataset.to_arrow_refss0  $ 2 2 4 4  5j\ B B  d33 ff % % ('F fbi ( (  )/::EEEE*EEEErzArgs: num_workerscp|#dttjz}t|||S)aConvert this dataset into a distributed RandomAccessDataset (EXPERIMENTAL). RandomAccessDataset partitions the dataset across the cluster by the given sort key, providing efficient random access to records via binary search. A number of worker actors are created, each of which has zero-copy access to the underlying sorted data blocks of the dataset. Note that the key must be unique in the dataset. If there are duplicate keys, an arbitrary value is returned. This is only supported for Arrow-format datasets. Args: key: The key column over which records can be queried. num_workers: The number of actors to use to serve random access queries. By default, this is determined by multiplying the number of Ray nodes in the cluster by four. As a rule of thumb, you can expect each worker to provide ~3000 records / second via ``get_async()``, and ~10000 records / second via ``multiget()``. N)r)rrnodesrh)rrrs rto_random_access_datasetz Dataset.to_random_access_dataset2s74  c#)++...K"4+FFFFrz store memory.)r insert_afterct|dt}|jj}fd|D}t t||j}tt|j |j|}| |j | |j|j|S)aExecute and materialize this dataset into object store memory. This can be used to read all blocks into memory. By default, Dataset doesn't read blocks from the datasource until the first transform. Note that this does not mutate the original Dataset. Only the blocks of the returned MaterializedDataset class are pinned in memory. Examples: >>> import ray >>> ds = ray.data.range(10) >>> materialized_ds = ds.materialize() >>> materialized_ds MaterializedDataset(num_blocks=..., num_rows=10, schema={id: int64}) Returns: A MaterializedDataset holding the materialized data blocks. T)rrc@g|]}t|gdjS)F)rr|rr})r block_with_metadatars rrz'Dataset.materialize..osH   $ +,!}      rrr)rrrurrrr,r2rr@rrrr _get_uuid)rrblocks_with_metadatarrrrs @rrzDataset.materializePs*||DT7J|KK J..00%}    (<    #9 #D#D#DdlSS $ $***,,4< H H H     """))***  rcR|jr=|jS|j7|jjr|jS|S)aReturns a string containing execution timing information. Note that this does not trigger execution, so if the dataset has not yet executed, an empty string is returned. Examples: .. testcode:: import ray ds = ray.data.range(10) assert ds.stats() == "" ds = ds.materialize() print(ds.stats()) .. testoutput:: :options: +MOCK Operator 0 Read: 1 tasks executed, 5 blocks produced in 0s * Remote wall time: 16.29us min, 7.29ms max, 1.21ms mean, 24.17ms total * Remote cpu time: 16.0us min, 2.54ms max, 810.45us mean, 16.21ms total * Peak heap memory usage (MiB): 137968.75 min, 142734.38 max, 139846 mean * Output num rows: 0 min, 1 max, 0 mean, 10 total * Output size bytes: 0 min, 8 max, 4 mean, 80 total * Tasks per node: 20 min, 20 max, 20 mean; 1 nodes used ) r get_stats to_summary to_stringrrhas_computed_outputr_get_stats_summaryrs rrz Dataset.statss>  ! *)3355@@BBLLNN N ^ 'DN,@,T,T,V,V '>'')) )&&((22444rcRt|jdS)aShow the logical plan and physical plan of the dataset. Examples: .. testcode:: import ray from ray.data import Dataset ds: Dataset = ray.data.range(10, override_num_blocks=10) ds = ds.map(lambda x: x + 1) ds.explain() .. testoutput:: -------- Logical Plan -------- MapRows[Map()] +- Read[ReadRange] -------- Logical Plan (Optimized) -------- MapRows[Map()] +- Read[ReadRange] -------- Physical Plan -------- TaskPoolMapOperator[Map()] +- TaskPoolMapOperator[ReadRange] +- InputDataBuffer[Input] -------- Physical Plan (Optimized) -------- TaskPoolMapOperator[ReadRange->Map()] +- InputDataBuffer[Input] N)rrexplainrs rrzDataset.explains'F dj  ""#####rcX|jSr)rrrrs rrzDataset._get_stats_summarys"z!!,,...rch|j\}}}||S)aGet an iterator over ``RefBundles`` belonging to this Dataset. Calling this function doesn't keep the data materialized in-memory. Examples: >>> import ray >>> ds = ray.data.range(1) >>> for ref_bundle in ds.iter_internal_ref_bundles(): ... for block_ref, block_md in ref_bundle.blocks: ... block = ray.get(block_ref) Returns: An iterator over this Dataset's ``RefBundles``. )rexecute_to_iteratorr)riter_ref_bundlesrs rroz!Dataset.iter_internal_ref_bundless7""&!?!?!A!A!Q &&(((rctd|jj}||S)aGet a list of references to the underlying blocks of this dataset. This function can be used for zero-copy access to the data. It blocks until the underlying blocks are computed. Examples: >>> import ray >>> ds = ray.data.range(1) >>> ds.get_internal_block_refs() [ObjectRef(...)] Returns: A list of references to this dataset's blocks. ze`Dataset.get_internal_block_refs()` is deprecated. Use `Dataset.iter_internal_ref_bundles()` instead.)rrrrrr)rrs rget_internal_block_refszDataset.get_internal_block_refssO"  =   Z''))4  &&(((rcltd|jjDS)aWhether this dataset's lineage is able to be serialized for storage and later deserialized, possibly on a different cluster. Only datasets that are created from data that we know will still exist at deserialization time, e.g. data external to this Ray cluster such as persistent cloud object stores, support lineage-based serialization. All of the ray.data.read_*() APIs support lineage-based serialization. Examples: >>> import ray >>> ray.data.from_items(list(range(10))).has_serializable_lineage() False >>> ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv").has_serializable_lineage() True c3>K|]}|VdSr)is_lineage_serializable)r r2s rr z3Dataset.has_serializable_lineage..sB    & & ( (      r)rrrpost_order_iterrs rhas_serializable_lineagez Dataset.has_serializable_lineagesD$  (,<<>>      rc|std|j}t j|jj}t ||}|j|| dtj j fd}tj jj} |tj j |t%j|}|tj j n)#|tj j wxYw|S)ak Serialize this dataset's lineage, not the actual data or the existing data futures, to bytes that can be stored and later deserialized, possibly on a different cluster. Note that this uses pickle and will drop all computed data, and that everything is recomputed from scratch after deserialization. Use :py:meth:`Dataset.deserialize_lineage` to deserialize the serialized bytes returned from this method into a Dataset. .. note:: Unioned and zipped datasets, produced by :py:meth`Dataset.union` and :py:meth:`Dataset.zip`, are not lineage-serializable. Examples: .. testcode:: import ray ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") serialized_ds = ds.serialize_lineage() ds = ray.data.Dataset.deserialize_lineage(serialized_ds) print(ds) .. testoutput:: Dataset(num_rows=?, schema=...) Returns: Serialized bytes containing the lineage of this dataset. aKLineage-based serialization is not supported for this stream, which means that it cannot be used as a tunable hyperparameter. Lineage-based serialization is explicitly NOT supported for unioned or zipped datasets (see docstrings for those methods), and is only supported for datasets created from data that we know will still exist at deserialization time, e.g. external data in persistent cloud object stores or in-memory data from long-lived clusters. Concretely, all ray.data.read_*() APIs should support lineage-based serialization, while all of the ray.data.from_*() APIs do not. To allow this stream to be serialized to storage, write the data to an external store (such as AWS S3, GCS, or Azure Blob Storage) using the Dataset.write_*() APIs, and serialize a new dataset reading from the external store using the ray.data.read_*() APIs.rfcF|\}}}d|d<|||fS)N_last_export_session_and_job) __reduce__)r reconstructorargsrs r_reduce_remote_fnz4Dataset.serialize_lineage.._reduce_remote_fnRs0*, &M448E0 1 $- -r)rrrrrrrclear_snapshotrrrremote_functionRemoteFunctionrworker global_workerget_serialization_context_register_cloudpickle_reducerpickledumps_unregister_cloudpickle_reducer)r plan_copylogical_plan_copyrrr serializeds rserialize_lineagezDataset.serialize_lineagesUH,,..  L "J((**  Idj&>?? Y 1 2 2 !!! T^^%%&&& .#"5"D . . . .,%3MMOO X  1 1#24E    b))J  3 3C4G4V W W W WG 3 3C4G4V W W W Ws )9E&E- serialized_dsc*tj|S)a Deserialize the provided lineage-serialized Dataset. This uses pickle, and assumes that the provided serialized bytes were serialized using :py:meth:`Dataset.serialize_lineage`. Examples: .. testcode:: import ray ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") serialized_ds = ds.serialize_lineage() ds = ray.data.Dataset.deserialize_lineage(serialized_ds) print(ds) .. testoutput:: Dataset(num_rows=?, schema=...) Args: serialized_ds: The serialized Dataset that we wish to deserialize. Returns: A deserialized ``Dataset`` instance. )rloads)rs rdeserialize_lineagezDataset.deserialize_lineagegs<|M***rc|jjS)z3Return the DataContext used to create this Dataset.)rrrs rrzDataset.contextsz""ragg_clsc:|j||g|Ri|}|j|S)aaHelper for aggregating on a particular subset of the dataset. This validates the `on` argument, and converts a list of column names or lambdas to a multi-aggregation. A null `on` results in a multi-aggregation on all columns for an Arrow Dataset, and a single aggregation on the entire row for a simple Dataset. )_build_multicolumn_aggsrU)rrr7rr_rSs rrOzDataset._aggregate_ons7,t+GRI$III&IIt~t$$r) skip_colsrcZ|\|d}|Dt|ts/sgt|jdkrfd|jD}t|t s|g}t|dkrt dfd|DS)zaBuild set of aggregations for applying a single aggregation to multiple columns. NTrHrcg|]}|v| Srr)r rxrs rrz3Dataset._build_multicolumn_aggs..s#NNN#I9M9M#9M9M9Mrz4At least 1 column to aggregate on has to be providedc(g|]}|gRiSrr)r on_rrr_s rrz3Dataset._build_multicolumn_aggs..s3<<<#-d---f--<<zz100%)width)layoutz text/plainr) rHTML __class__r@ _tab_repr_VBoxLayout_repr_mimebundle_updaterepr)rr_rtitletabwidgetrs rrzDataset._repr_mimebundle_s"  Et~'> E E EFFoo%j6G6Gf6G6U6UVV*)33F33 d4jj     rc ddlm}m}|j|d}|d}|$tdd}nt|trGtdd tj t|d }ni}t|j|jD]&\}}t#|d t|||<'td t%|d dddgd}g} | |td t%|d dddgd| |||| ddgS)Nr)rTab)rMr FrHzrendered_html_common.html.j2z
Unknown schema
)contentz
Data type: z
r@zscrollableTable.html.j2htmlNameType) tabular_datatablefmt showindexheaders300px)r max_heightFieldValueMetadatar)titles)rrrrrjrrrnrenderrrrescaper r-rrlgetattrrr,r) rrrrr schema_repr schema_datasnamestypechildrens rrzDataset._tab_repr_s;((((((((*7799((**   e44 >"#ABBII1JKK % % "#ABBIIV CKK0H0HVVVJKKK #FL&, ? ? L L u%,UJE %K%K E"""#<==DD!,!2!2!4!4###V,  #EK D233::"%-^^%5%5!'"'!(' 2  ';    [))***s8Z$:;;;;rc@|j|jSr)rget_plan_as_stringrrs r__repr__zDataset.__repr__sz,,T^<< >???? @wx   rc4|jSr)rrrs rrzDataset._meta_count4sz$$&&&rc|jSr)rqrs rrzDataset._get_uuid7s zruuidcN||_||j_||jj_dSr)rqr _dataset_uuid _in_statsr)rr"s rrzDataset._set_uuid:s& #'  ,0 )))rc\|jr$|jdd|_dSdS)aFlush progress bar output by shutting down the current executor. This should be called at the end of all blocking APIs (e.g., `take`), but not async APIs (e.g., `iter_batches`). The streaming executor runs in a separate generator / thread, so it is possible the shutdown logic runs even after a call to retrieve rows from the stream has finished. Explicit shutdown avoids this, which can clobber console output (https://github.com/ray-project/ray/issues/32414). TforceN)rshutdownrs rrz!Dataset._synchronize_progress_bar?sC  ! *  " + +$ + 7 7 7%)D " " "  * *rcR|j\}}}||_||fSr)rrr)r bundle_iterrexecutors rr zDataset._execute_to_iteratorPs1'+z'E'E'G'G$ UH"*E!!rc,|j|j|jdS)N)rr"r)rrqrrs r __getstate__zDataset.__getstate__Xs#JJ .   rcb|d|_|d|_|d|_d|_dS)Nrr"r)rrqrr)rrs r __setstate__zDataset.__setstate__`s36] 6] ">2!%rc|jsdS t0tjr|jddSdSdS#t$rYdSwxYw)NFr')rris_initializedr)rrs r__del__zDataset.__del__fsz%  F  3#5#7#7&//e/<<<<<      DD s5A AA)FN)NN)r0NNNr)Fr)NrT)r)T)r)rTNN)rrI)rrv)rr)rrrr)rru)r@ __module__ __qualname____doc__r@r,r staticmethodboolr rrrl BT_API_GROUPrrr rrrrrrrrrjrrpropertyrrrr\rXr rrEXPRESSION_API_GROUPrvrrYrr rrr/r3rI SSR_API_GROUPrQrHrTrWrirI SMJ_API_GROUPrgrtrrrrrrrr$r/rF GGA_API_GROUPrrRrMrUr=rZr\r^rarfrNrzrLr-r CD_API_GROUPrrerr IM_API_GROUPrrrbrMrr IOC_API_GROUPrbAPPENDrarrrrrrrrrrr_rrrr rCREATErrrr`rr-rr{rTorchDeviceTyperrTorchBatchTyper=TensorFlowTensorBatchTyper@rHrMrurrrrnrkrirrrrhr E_API_GROUPrrrrFrr r&rorVrrbytesrrr^rrOrr r[rProrrrrrrrrrrrrrEr r.r0r3rrrrrs5GGRHH"HHHH.GK:: :#':6>tn: :::\:Y&&& .2+/.27;:>$($("&SWEIP+P+P+ d38n%tCH~5 6P+/* P+ (3-( P+ DcN+ P+&hsm4P+ (S#X7P+5/P+5/P+P+eCsCx%S# :N$NOPP+%Xb$sCx..@%ABP+ P+P+P+'&P+dZ0$???hsm@?(Xc]((((Z,d;;;x}<;X(hsm(((X(+++++ Y&&& <@-1&/ $+/.27;:>$($("&SW(-EI#F F F  9 4 5F #tWY%778 F /* F sm F F (3-(F DcN+F &hsm4F  (S#X7F 5/F 5/F F eCsCx%S# :N$NOPF "&!F "%Xb$sCx..@%AB#F & 'F F F '&F PK+  9 4 5K+#tWY%778 K+ /* K+ sm K+K+(3-(K+DcN+K+&hsm4K+ (S#X7K+5/K+5/K+K+eCsCx%S# :N$NOPK+ "&!K+"%Xb$sCx..@%AB#K+K+K+K+ZY-AAAC+C+C+  C+C+C+BAC+JZ7888Y&&&'/!%%)q q q q   K   q smq #q c]q  q q q '&98q fY&&& "&%) 7 7 7 3i7 # 7 c] 7  7 7 7 '&7 rY&&& 04%) N+N+N+CcN#N+sO+, N+ c] N+ N+N+N+'&N+`Y&&& TX {+{+{+T#YS#X./{+eCsCx%S# :N$NOP {+{+{+'&{+zY&&&.2+/.27;:>$($("&SWEIJ+J+J+  cNE$tCH~"6S#X"FG G J+ /* J+(3-(J+DcN+J+&hsm4J+ (S#X7J+5/J+5/J+J+eCsCx%S# :N$NOPJ+%Xb$sCx..@%ABJ+" #J+J+J+'&J+XY&&&CG+/D+ 04+/.27;:>$($("&SWEID+D+D+ (c3h)=> ?D+uS$Y'(D+ sO+, D+ (3-( D+DcN+D+&hsm4D+ (S#X7D+5/D+5/D+D+eCsCx%S# :N$NOPD+%Xb$sCx..@%ABD+ !D+D+D+'&D+LY'''%)37z+ $(z+z+z+SMz+$,C=z+  z+ tCy! z+z+ z+z+z+('z+xY'''#$( .+.+.+sm.+SM .+  .+.+.+('[.+`Y'''#!+!+!+sm!+  !+!+!+('[!+FY&&&8<@ @ @ @ (0 @ @ @ @ '&@ DY''' 6: ePePeP eP eP !k!23 eP l  ePePeP('^ePNY'''',TX```` $`>FtCy>Q` # $```('^`DY'''KS Kd;P6QKKK('^KZY'''M4;M4 # $M4M4M4('^M4^Y''' ""& C@C@C@e$C@ C@ sm C@ 3- C@ ; <C@C@C@('^C@J%!%!(-c5j(9%!EH%! ; <%!%!%!%!N 5 T    ),YM::: 19%)" N!N!N!N!,- N! c] N! sm N! # $N!N!N!;:N!`Y'''+ DO+ + + + ('+ ZY''' !)-%)&*l<.2?C!&l<l<l< l<l< l< #J l< 5:& l<c]l<sml<&c]l<%-T#s(^$<l<l< l<l<l<('[l<\Y''')-:E:E 3S 4' (:E! :E  :E:E:E('[:ExY'''(+(+S(+(+c(+(+(+('^[(+TY'''$0{$0uS$sCx.5H/I$0$0$0('^[$0LY'''OS'+'+5d3i01'+HL'+ sDcN" #'+'+'+('^['+RY'''OS'+'+5d3i01'+HL'+ sDcN" #'+'+'+('^['+RY'''OS'+'+5d3i01'+HL'+ sDcN" #'+'+'+('^['+RY'''OS'+'+5d3i01'+HL'+ sDcN" #'+'+'+('^['+RY'''/3! 6+6+ U3S >* +6+6+ 6+ sDcN" # 6+6+6+('^[6+pY':::(,  D D $s)$D %- 8SE4 +>$>?? @% D  D D D ;:^[D LY'''/4.2 D+D+ 3S > "D+$T *+D+sEz*+ D+  D+D+D+('[D+LY''''+$y/'+i'+'+'+(''+RY&&&+3+9+++'&+4Y&&& "8EN88885=c]8 888'&^8tY&&&//#/tDcN';///'&^/bY&&&&&hsm&tDcN7K&&&'&^&PY&&&#t'&^6^' Y&&&*s***'&  *X^$G"  Y&&&&&t&x7I&&&'& &P^$G"  Y&&&c8K'& 6Y&&& C   '& $Y&&&3C333'&^3,Y&&& 3T#Y 3 3 3'&^ 3Y''' /38<#;?8/A&BCz $C=z c3hz c]z $C=z z  z z z ('^z xYM:::488<#?-12659*.%)q q q !c3h0q &d38n5 q  q #6* q  S#X/q #4S>2q c3hq c]q  q q q ;:^q fYM::: ! J 9=#;?8<*.%)!J J J J J  J 45 J J !)c3h 8J $$45J c3hJ c]J J  J J J ^;:J XY''' 9=#;?8B%)+/!^ ^ ^ ^ 45 ^  ^ !)c3h 8 ^ $$45^ $C=^ c3h^ %c8T 9:;^ c]^ $C=^ ^  ^ ^ ^ ;:^^ @Y''' 9=#;?8<+/*.%)+/!a a a a  a 45 a  a !)c3h 8a $$45a $C=a c3ha c]a $C=a a  a a a ('^a F 59%) A A A %R^4A "$sCx.1 A c] A  A A A ^A F +/%) 8 8 8 8  #8 c3h 8 c] 8 8 8 ^8 tYM::: +/%) G G G G  G c3h G c] G  G G G ^;:G R  *.*.%)G G G G  G "$ G c3h G c]G  G G G ^G R "-14826<@/3*.%)B B B B B  B )* B "$sCx.1B  S#X/B !!89B  (}B c3hB c]B  B B B ^B H .29A!,!1.248*.%)2 2 2 2 )* 2 56 2  2 2 'sm2 "$sCx.12 c3h2 c]2  2 2 2 ^2 h^./// +/%) NNNNc3h N c] N  NNN0/N`^ U  Y&&&&,&&&'& &Y&&&+8DcN3+++'&^+$Y&&&!"$'&/37,0EIC C C C SM C sm C  C $,C=C %SMC h {L'@ABC  ) C C C '&^C JY&&&!"$'KO:@PT37,0g g g g SM g }d3 3E.FFGH g owv67 g XtCO'<&=|&KLMg g $,C=g %SMg  . !g g g '&^g R!"$'SW37,0T T T T SM T 0$s?&F&F&F\0/&FP^G$$$&*GG Gc]G  GGG%$G:^O$???Y%%%///&%@?/bY&&&"5s"5"5"5'&"5HY999"$"$:9"$H/$7////^K((( 8I+>   \)( (^K(((i.>)?)(Z. $   \ ,N5NNN\N`+5+Y+++\\+<####\X# % %!)%T#Y*?!@ % % % %$*. ==== U3S >* += DI& ====8 eWn(= !    s+,,-,@/</</>> import ray >>> ds = ray.data.range(100).repartition(10).materialize() >>> ds.num_blocks() 10 Time complexity: O(1) Returns: The number of blocks of this :class:`Dataset`. )rrjrs rrMzMaterializedDataset.num_blockssz,,...rN)r@r4r5r6rrMrrrruruws9/C//////rrubeta)rceZdZdZdddeddeefdZede e fd Z ede ee e d ffd Zd Zd ZdS)rzaDataset schema. Attributes: base_schema: The underlying Arrow or Pandas schema. Nrro)zpyarrow.lib.Schemar?rcl||_|p$tjtj|_dS)z Initialize a :class:`Schema` wrapper around an Arrow or Pandas schema. Args: base_schema: The underlying Arrow or Pandas schema. data_context: The data context to use for this schema. N)rordeepcopyr^r^r)rrors rrzSchema.__init__s1'%P k6M6O6O(P(P rrc|jjS)z"Lists the columns of this Dataset.)rorrs rrz Schema.namess%%rzpyarrow.lib.DataTypec  ddl ddlddlmddlm}m}dttj j fdj f fd }t|j jjrt!|j jSg}|j jD]}t||rJ|jjrt(}n|}|||j||j\ ||||#j$r|t2Yt4$r6t6d |d |dYwxYw|S) zuLists the types of this Dataset in Arrow format For non-Arrow compatible types, we return "object". rN)BaseMaskedDtype)ArrowTensorTyperYr\rct|jr|jSt|jrSt|r|j}|Sr)r ArrowDtype pyarrow_dtype StringDtypestring numpy_dtypefrom_numpy_dtype)r\rQrrs r_convert_to_pa_typez)Schema.types.._convert_to_pa_typesr%// ***E2>22 *yy{{"E?33 *)&&u-- -r)shaper\zError converting dtype z to Arrow.)rrpandas.core.dtypes.dtypesrQrkrRrYrrr\rTrfrrolibrrrlruse_arrow_tensor_v2rr_shape_dtypeArrowNotImplementedErrorobjectr$r exception) rrRrYrZ arrow_typesr\pa_tensor_type_classrQrrs @@@rrlz Schema.typess ======DDDDDDDD .2=/AB . [ . . . . . . . . d& 6 6 0(.// / %+ - -E%-- -=4;+<((+:(""((#l11%,??-&&':':5'A'ABBBB2///&&v..... ---$$%Pu%P%P%PQQQ&&t,,,,,-s+D  $E/0.s===$CII===rColumnr9 zType -r )r\rrr-rlrstrip)r column_widthpaddingrrrs rrzSchema.__repr__s==$*===XOPP #,0CMMABB(#H %%#,0CMMABB#F #d**dj$*55 " "JD$ dNF clW4D AB BF kkk !FF r)r@r4r5r6rr r^rr:r r rrrbrlrgrrrrrrs/3 QQQDEQ{+ QQQQ&&tCy&&&X&2tE$v,0F"FGH222X2h   rrrrrOcRtj|}|Sr)rWrrnrs rrWrWs"  #E * *E ??  rrcTtj|}||Sr)rWrr)rrs rrrs$  #E * *E >>& ! !!rcRtj|}|Sr)rWrto_arrowrrs rrr s"  #E * *E >>  r)rrrrloggingrrtypingrrrrrrr r r r r rrrrrrray.cloudpickle cloudpicklerray._common.usager)ray._private.thirdparty.tabulate.tabulater$ray.air.util.tensor_extensions.arrowrrrr/ray.data._internal.datasource.bigquery_datasinkr1ray.data._internal.datasource.clickhouse_datasinkrrr*ray.data._internal.datasource.csv_datasinkr.ray.data._internal.datasource.iceberg_datasinkr,ray.data._internal.datasource.image_datasinkr+ray.data._internal.datasource.json_datasinkr,ray.data._internal.datasource.lance_datasinkr,ray.data._internal.datasource.mongo_datasinkr,ray.data._internal.datasource.numpy_datasinkr .ray.data._internal.datasource.parquet_datasinkr!*ray.data._internal.datasource.sql_datasinkr"0ray.data._internal.datasource.tfrecords_datasinkr#1ray.data._internal.datasource.webdataset_datasinkr$ray.data._internal.equalizer%'ray.data._internal.execution.interfacesr&2ray.data._internal.execution.interfaces.ref_bundler'!ray.data._internal.execution.utilr()ray.data._internal.iterator.iterator_implr*1ray.data._internal.iterator.stream_split_iteratorr+%ray.data._internal.logical.interfacesr,8ray.data._internal.logical.operators.all_to_all_operatorr-r.r/r03ray.data._internal.logical.operators.count_operatorr18ray.data._internal.logical.operators.input_data_operatorr22ray.data._internal.logical.operators.join_operatorr3rr4r5r6r7r8r93ray.data._internal.logical.operators.n_ary_operatorr,r:rr;=ray.data._internal.logical.operators.streaming_split_operatorr<3ray.data._internal.logical.operators.write_operatorr=rfr>r?ray.data._internal.planr@2ray.data._internal.planner.exchange.sort_task_specrAray.data._internal.remote_fnrBray.data._internal.splitrCrDray.data._internal.statsrErFrGray.data._internal.utilrHrIrJrKrLray.data.aggregaterMrNrOrPrQrRrSrTray.data.blockrUrVrWrXrYrZr[r\r]ray.data.contextr^ray.data.datasourcer_r`rarbray.data.datasource.datasinkrcrd!ray.data.datasource.file_datasinkreray.data.datatyperfray.data.iteratorrgray.data.random_access_datasetrh ray.typesriray.util.annotationsrjrkrlray.util.scheduling_strategiesrm ray.widgetsrnray.widgets.utilrorKrcmarsmodinrrpyspark tensorflowtftorchtorch.utils.datatensorflow_metadata.proto.v0rprqrrrJrtrmrurrvrwrx getLoggerr@rrr TensorflowFeatureTypeSpecrFr{rErrDr9r<r=r>r?rAr@rGr;rrurrWrrrrrrs   " ''''''>>>>>>766666LLLLLL CBBBBBJJJJJJFFFFFFDDDDDDFFFFFFFFFFFFFFFFFFJJJJJJBBBBBBMMMMMMPPPPPP111111======<;;;;;FFFFFFUUUUUU====== FEEEEENNNNNNCCCCCCKJJJJJXXXXXXEEEEEEQQQQQQQQ111111FFFFFF999999EEEEEEEEUUUUUUUUUU                                          )(((((PPPPPPPPPPPPPPPPPPPP;;;;;;&&&&&&******>>>>>>DDDDDDDDDDIIIIII //////.KKKKKKKKKLLLMMMNNNNNNLLL777777KKKKKKKK111111------4444444444  8 $ $?!4 &S--?(@@"+tC4D/E"EFw~&& tC/0,>?^S01 ' 7 6 1  $ $  $ |f|f|f|f|f|f|f |f~M /////'71:// /2 Vmmmmmmmm`"4 "U"HSM"""" 5r