&`iddlZddlZddlZddlmZmZmZmZmZm Z m Z m Z m Z m Z mZmZddlZddlmZddlmZddlmZddlmZddlmZddlmZdd lmZm Z m!Z!dd l"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*dd l+m,Z,dd l-m.Z.m/Z/erddl0Z1ddl2Z2dd l3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9edZ:Gddee:Z;e.Gddej<Z=e=Z>dS)N) TYPE_CHECKINGAnyCallableDictIterableIteratorListLiteralOptionalTupleTypeVarUnion BatchIterator) RefBundle) LogicalPlan) InputData) ExecutionPlan) DatasetStats) BlockAccessor DataBatch_apply_batch_format)ArrowBatchCollateFn CollateFnDefaultCollateFnNumpyBatchCollateFnPandasBatchCollateFnTensorBatchReturnTypeTensorBatchTypeis_tensor_batch_type) DataContext) PublicAPIRayDeprecationWarning) CollatedDataMaterializedDatasetSchemaTensorFlowTensorBatchTypeTorchBatchTypeTorchDeviceTypeTc<eZdZdegeeffdZdZdS)_IterableFromIterator iterator_genc||_dS)zConstructs an Iterable from an iterator generator. Args: iterator_gen: A function that returns an iterator each time it is called. For example, this can be a generator function. Nr-)selfr-s e/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/iterator.py__init__z_IterableFromIterator.__init__;s)c*|SNr/r0s r1__iter__z_IterableFromIterator.__iter__Ds  """r3N)__name__ __module__ __qualname__rrr*r2r7r3r1r,r,:sK)Xb(1+o%>))))#####r3r,ceZdZdZejdeeee e e ffdZ e ddddddd d ed ed e ed e de ede edeefdZdeedefdZdddddddddd ed ed e ed e de ede ede eegdfde eegefdeefdZdefdZe deeeeffdZeje defdZejdAdZejdefdZe dddddddddd d ed e ed e ed!eed!ffd"ed#e dfd$e eeeee!j"fgdfe#fd e de ede ed%e ded&fd'Z$ddddddd(d ed e ed e ed)eed)ffd e de ede eded*fd+Z%dddddddddd,d,d- d.e ed/e ee&ee&e&eeee&effd0e d!d1e ed!e&d!eed!ffd ed ed e de ede ed2e d3e dd4fd5Z'e dddddddddd6 d/eee&efd7eee&efd8ee ee e&efd ed ed e de ede ed9ed:eed:ffd;ed:eed:ffdZ(e dBd@Z)dS)C DataIteratora(An iterator for reading records from a :class:`~Dataset`. For Datasets, each iteration call represents a complete read of all items in the Dataset. If using Ray Train, each trainer actor should get its own iterator by calling :meth:`ray.train.get_dataset_shard("train") `. Examples: >>> import ray >>> ds = ray.data.range(5) >>> ds Dataset(num_rows=5, schema={id: int64}) >>> ds.iterator() DataIterator(Dataset(num_rows=5, schema={id: int64})) returncdS)aReturns the iterator to use for `iter_batches`. Returns: A tuple. The first item of the tuple is an iterator over RefBundles. The second item of the tuple is a DatasetStats object used for recording stats during iteration. The third item is a boolean indicating if the blocks can be safely cleared after use. Nr;r6s r1_to_ref_bundle_iteratorz$DataIterator._to_ref_bundle_iterator\s  r3defaultFNprefetch_batches batch_size batch_format drop_lastlocal_shuffle_buffer_sizelocal_shuffle_seedrErFrGrHrIrJc8|||||||S)aReturn a batched iterable over the dataset. Examples: >>> import ray >>> for batch in ray.data.range( ... 1000000 ... ).iterator().iter_batches(): # doctest: +SKIP ... print(batch) # doctest: +SKIP 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 will be 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. batch_format: Specify ``"default"`` to use the default block format (NumPy), ``"pandas"`` to select ``pandas.DataFrame``, "pyarrow" to select ``pyarrow.Table``, or ``"numpy"`` to select ``Dict[str, numpy.ndarray]``, or None to return the underlying block exactly as is with no additional formatting. drop_last: Whether to drop the last batch if it's incomplete. local_shuffle_buffer_size: If non-None, the data will be 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 will be drained. local_shuffle_seed: The seed to use for the local random shuffle. Returns: An iterable over record batches. rD) _iter_batches)r0rErFrGrHrIrJs r1 iter_batcheszDataIterator.iter_batchesks4^!!-!%&?1 "   r3ref_bundles_iterc t|fi|Sr5r)r0rNkwargss r1_create_batch_iteratorz#DataIterator._create_batch_iterators-88888r3rErFrGrHrIrJ _collate_fn _finalize_fnrSr$rTc tdttff d } t| S)Nr>c3 Ktj}\}}}}||||    }|r.|jtj|z |Ed{V|r0|jtj|z dSdS)N) stats dataset_tagclear_block_after_readrFrGrH collate_fn finalize_fnshuffle_buffer_min_size shuffle_seedrE)time perf_counterr@_get_dataset_tagrQiter_initialize_sadd iter_total_s) time_startref_bundles_iteratorrWblocks_owned_by_consumerrXbatch_iteratorrSrTrGrFrHrIrJrEr0s r1_create_iteratorz4DataIterator._iter_batches.._create_iterators *,,J,,..  $(//11K!88$''?%)#&((A/!19  N N'++D,=,?,?*,LMMM% % % % % % % % I"&&t'8':':Z'GHHHHH I Ir3)rrrr,) r0rErFrGrHrIrJrSrTrhs ````````` r1rLzDataIterator._iter_batchess+<88 $ I(9"5$ I$ I$ I$ I$ I$ I$ I$ I$ I$ I$ I$ I$ I$ IL%%5666r3cdS)Nunknown_datasetr;r6s r1r`zDataIterator._get_dataset_tags  r3c\|dddfd}t|S)aReturn a local row iterable over the dataset. If the dataset is a tabular dataset (Arrow/Pandas blocks), dicts are yielded for each row by the iterator. If the dataset is not tabular, the raw row is yielded. Examples: >>> import ray >>> dataset = ray.data.range(10) >>> next(iter(dataset.iterator().iter_rows())) {'id': 0} Time complexity: O(1) Returns: An iterable over rows of the dataset. NrA)rFrGrEc3KD]E}tjtj|}|dD]}|VFdS)NT)public_row_format)r for_blockbatch_to_block iter_rows)batchrowbatch_iterables r1_wrapped_iteratorz1DataIterator.iter_rows.._wrapped_iteratorsh'  %/ 0LU0S0STT ??T?BBCIIII  r3)rLr,)r0rtrss @r1rpzDataIterator.iter_rowssP&++$,        %%6777r3cdS)z9Returns a string containing execution timing information.Nr;r6s r1rWzDataIterator.statss  r3r&cdS)z/Return the schema of the dataset iterated over.Nr;r6s r1schemazDataIterator.schemas  r3cdSr5r;r6s r1 get_contextzDataIterator.get_context s r3auto) rErFdtypesdevicerZrHrIrJ pin_memoryr{z torch.dtyper|r)rZr}r(c ddlm} ddlm} ||dkrt d| r|t ddkr| r | nddd lmd td tttfffd } |t|| }d} nt|trd} n{t|trd} nct|trd} nKt!|rd} t#jdt&nt dt)||||| ||||| S)aReturn a batched iterable of Torch Tensors over the dataset. This iterable yields a dictionary of column-tensors. If you are looking for more flexibility in the tensor conversion (e.g. casting dtypes) or the batch format, try using :meth:`~ray.data.DataIterator.iter_batches` directly. Examples: >>> import ray >>> for batch in ray.data.range( ... 12, ... ).iterator().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 ``ArrowBatchCollateFn`` to customize how the tensor batch is created from an Arrow batch. >>> import pyarrow as pa >>> import torch >>> import ray >>> from ray.data.collate_fn import ArrowBatchCollateFn >>> class CustomArrowBatchCollateFn(ArrowBatchCollateFn): ... def __call__(self, batch: pa.Table) -> torch.Tensor: ... return torch.as_tensor(batch["col_1"].to_numpy() + 5) >>> iterator = ray.data.from_items([ ... {"col_1": 1, "col_2": 2}, ... {"col_1": 3, "col_2": 4}]).iterator() >>> for batch in iterator.iter_torch_batches(collate_fn=CustomArrowBatchCollateFn()): ... print(batch) tensor([6, 8]) Use the ``NumpyBatchCollateFn`` to customize how the tensor batch is created from a Numpy batch. >>> from typing import Dict >>> import numpy as np >>> import torch >>> import ray >>> from ray.data.collate_fn import NumpyBatchCollateFn >>> class CustomNumpyBatchCollateFn(NumpyBatchCollateFn): ... def __call__(self, batch: Dict[str, np.ndarray]) -> torch.Tensor: ... return torch.as_tensor(batch["col_1"] + 5) >>> iterator = ray.data.from_items([ ... {"col_1": 1, "col_2": 2}, ... {"col_1": 3, "col_2": 4}]).iterator() >>> for batch in iterator.iter_torch_batches(collate_fn=CustomNumpyBatchCollateFn()): ... print(batch) tensor([6, 8]) Use the ``PandasBatchCollateFn`` to customize how the tensor batch is created from a Pandas batch. >>> import pandas as pd >>> import torch >>> import ray >>> from ray.data.collate_fn import PandasBatchCollateFn >>> class CustomPandasBatchCollateFn(PandasBatchCollateFn): ... def __call__(self, batch: pd.DataFrame) -> torch.Tensor: ... return torch.as_tensor(batch["col_1"].to_numpy() + 5) >>> iterator = ray.data.from_items([ ... {"col_1": 1, "col_2": 2}, ... {"col_1": 3, "col_2": 4}]).iterator() >>> for batch in iterator.iter_torch_batches(collate_fn=CustomPandasBatchCollateFn()): ... print(batch) tensor([6, 8]) 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 will be 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 will be 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: [Alpha] A function to customize how data batches are collated before being passed to the model. This is useful for last-mile data formatting such as padding, masking, or packaging tensors into custom data structures. If not provided, `iter_torch_batches` automatically converts batches to `torch.Tensor`s and moves them to the device assigned to the current worker. The input to `collate_fn` may be: 1. pyarrow.Table, where you should provide a callable class that subclasses `ArrowBatchCollateFn` (recommended for best performance). Note that you should use util function `arrow_batch_to_tensors` to convert the pyarrow.Table to a dictionary of non-contiguous tensor batches. 2. Dict[str, np.ndarray], where you should provide a callable class that subclasses `NumpyBatchCollateFn` 3. pd.DataFrame, where you should provide a callable class that subclasses `PandasBatchCollateFn` The output can be any type. If the output is a `TensorBatchType`, it will be automatically moved to the current worker's device. For other types, you must handle device transfer manually in your training loop. Note: This function is called in a multi-threaded context; avoid using thread-unsafe code. drop_last: Whether to drop the last batch if it's incomplete. local_shuffle_buffer_size: If non-None, the data will be 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 will be 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. pin_memory: [Alpha] If True, copies the tensor to pinned memory. Note that `pin_memory` is only supported when using `DefaultCollateFn`. Returns: An iterable over Torch Tensor batches. r) get_device)_in_ray_train_workerNrzzcollate_fn cannot be used with dtypes and device.You should manually move the output Torch tensors to thedesired dtype and device outside of collate_fn.z;pin_memory is only supported when using `DefaultCollateFn`.cpu)move_tensors_to_devicerqr>c@t|r |S|S)aDefault finalize function for moving PyTorch tensors to device. If batch is of type `TensorBatchType`, it will be automatically moved to the current worker's device. For other types, you must handle device transfer manually in your training loop. Args: batch: Input batch to move to device. Returns: Batch with tensors moved to the target device. - If input is TensorBatchType, returns tensors moved to device - Otherwise returns the same type as input without moving tensors to device. )r|)r )rqr|rs r1default_finalize_fnz.default_finalize_fns0"$E** --eFCCCC r3)r{r|r}pyarrownumpypandaszPassing a function to `iter_torch_batches(collate_fn)` is deprecated in Ray 2.47. Please switch to using a callable class that inherits from `ArrowBatchCollateFn`, `NumpyBatchCollateFn`, or `PandasBatchCollateFn`.zUnsupported collate function: rR)ray.train.torchrray.train.utilsr ValueErrorray.air._internal.torch_utilsrrrrrr isinstancerrrcallablewarningswarnr#typerL)r0rErFr{r|rZrHrIrJr}rrrrGrs ` @r1iter_torch_batcheszDataIterator.iter_torch_batchess\ /.....888888  !v'9Vv=M=MB   *0M  V  &:%9%;%;FZZ\\\F       "  (#- .       ,  *%J %LL  $7 8 8 R%LL  $7 8 8 R"LL  $8 9 9 R#LL j ! ! R"L M*&     Pd:>N>NPPQQ Q!!-!%&?1","   r3)rErFr{rHrIrJztf.dtypes.DTyper'cr ddlm ||||||}t fd|}|S)a\ Return a batched iterable of TensorFlow Tensors over the dataset. This iterable will yield single-tensor batches of the underlying dataset consists of a single column; otherwise, it will yield a dictionary of column-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: >>> import ray >>> for batch in ray.data.range( # doctest: +SKIP ... 12, ... ).iter_tf_batches(batch_size=4): ... print(batch.shape) # doctest: +SKIP (4, 1) (4, 1) (4, 1) 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 will be 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 TensorFlow dtype(s) for the created tensor(s); if None, the dtype will be inferred from the tensor data. drop_last: Whether to drop the last batch if it's incomplete. local_shuffle_buffer_size: If non-None, the data will be 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 will be 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. Returns: An iterator over TensorFlow Tensor batches. r)(convert_ndarray_batch_to_tf_tensor_batchrErFrHrIrJc|S)N)r{r;)rqrr{s r1z.DataIterator.iter_tf_batches..>sBBfr3)"ray.air._internal.tensorflow_utilsrrLmap) r0rErFr{rHrIrJrsmapped_iterablers ` @r1iter_tf_batcheszDataIterator.iter_tf_batchesst      ++-!&?1 ,           r3T) label_columnfeature_columnslabel_column_dtypefeature_column_dtypesrFrErHrIrJunsqueeze_label_tensorunsqueeze_feature_tensorsrrrrrrz torch.utils.data.IterableDatasetc  ddl} ddlmddlm} sdrt | jsit trt ts tdtdttkrtdtd Drtd nt dttfrt ttfs td tdt!t!krtd td Drtd   f d }| |S)ahReturn a Torch IterableDataset over this dataset. This is only supported for datasets convertible to Arrow records. It is recommended to use the returned ``IterableDataset`` directly instead of passing it into a torch ``DataLoader``. Each element in IterableDataset will be a tuple consisting of 2 elements. The first item contains the feature tensor(s), and the second item is the label tensor. Those can take on different forms, depending on the specified arguments. For the features tensor (N is the ``batch_size`` and n, m, k are the number of features per tensor): * If ``feature_columns`` is a ``List[str]``, the features will be a tensor of shape (N, n), with columns corresponding to ``feature_columns`` * If ``feature_columns`` is a ``List[List[str]]``, the features will be a list of tensors of shape [(N, m),...,(N, k)], with columns of each tensor corresponding to the elements of ``feature_columns`` * If ``feature_columns`` is a ``Dict[str, List[str]]``, the features will be a dict of key-tensor pairs of shape {key1: (N, m),..., keyN: (N, k)}, with columns of each tensor corresponding to the value of ``feature_columns`` under the key. If ``unsqueeze_label_tensor=True`` (default), the label tensor will be of shape (N, 1). Otherwise, it will be of shape (N,). If ``label_column`` is specified as ``None``, then no column from the ``Dataset`` will be treated as the label, and the output label tensor will be ``None``. Note that you probably want to call ``.split()`` on this dataset if there are to be multiple Torch workers consuming the data. Time complexity: O(1) Args: label_column: The name of the column used as the label (second element of the output list). Can be None for prediction, in which case the second element of returned tuple will also be None. feature_columns: The names of the columns to use as the features. Can be a list of lists or a dict of string-list pairs for multi-tensor output. If None, then use all columns except the label column as the features. label_column_dtype: The torch dtype to use for the label column. If None, then automatically infer the dtype. feature_column_dtypes: The dtypes to use for the feature tensors. This should match the format of ``feature_columns``, or be a single dtype, in which case it will be applied to all tensors. If None, then automatically infer the dtype. batch_size: How many samples per batch to yield at a time. Defaults to 1. prefetch_batches: The number of batches to fetch ahead of the current batch to fetch. If set to greater than 0, a separate threadpool will be used to fetch the objects to the local node, format the batches, and apply the collate_fn. 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 dataset is not divisible by the batch size, then the last batch will be smaller. Defaults to False. local_shuffle_buffer_size: If non-None, the data will be 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 will be 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. unsqueeze_label_tensor: If set to True, the label tensor will be unsqueezed (reshaped to (N, 1)). Otherwise, it will be left as is, that is (N, ). In general, regression loss functions expect an unsqueezed tensor, while classification loss functions expect a squeezed one. Defaults to True. unsqueeze_feature_tensors: If set to True, the features tensors will be unsqueezed (reshaped to (N, 1)) before being concatenated into the final features tensor. Otherwise, they will be left as is, that is (N, ). Defaults to True. Returns: A torch IterableDataset. rN)convert_pandas_to_torch_tensor)TorchIterableDatasetzbIf `feature_columns` is a dict, `feature_column_dtypes` must be None, `torch.dtype`, or dict, got .zF`feature_columns` and `feature_column_dtypes` must have the same keys.c3K|]}| VdSr5r;.0 subcolumnss r1 z(DataIterator.to_torch..s$QQ*:~QQQQQQr3zcolumn list may not be emptyzqIf `feature_columns` is a list of lists, `feature_column_dtypes` must be None, `torch.dtype`, or a sequence, got zH`feature_columns` and `feature_column_dtypes` must have the same length.c3K|]}| VdSr5r;rs r1rz(DataIterator.to_torch..s$HH*:~HHHHHHr3c3 K d  D]ir&g }nd}ttrfdD}n}||fVjdS)Nr)rFrGrErHrIrJ unsqueezec vi|]5}||ttr|n6S)r)rdict)rkeyrqrrrrs r1 zADataIterator.to_torch..make_generator..sq ' ' ' ;;!+C0$..CT#J#J!; 5c : :%:&?    ' ' 'r3)columns column_dtypesr)rLpoprr) label_tensorfeatures_tensorrqrFrrHrrrrrIrJrEr0rrs @r1make_generatorz-DataIterator.to_torch..make_generators"++%%!1#*C#5 ,) 6) 6 (#A#A%*"8 $$$L IIl++++#'Lot44 ' ' ' ' ' ' ' '$3 ' ' 'OO'E&D /&;"; '''O' 55555S) 6) 6r3)torchrr)ray.data._internal.torch_iterable_datasetrrdtyper TypeErrorrsetranyvalueslisttuplelen)r0rrrrrFrErHrIrJrrrrrrs```````````` @r1to_torchzDataIterator.to_torchFsxX  PPPPPPRRRRRR #"O  E4I5;)W)W E/400 E!"7>>#H)-.C)D)DHHH ''3/D+E+EEE$3QQ8N8N8P8PQQQQQE$%CDDDEOA.u >> E!"7$GG#N/34I/J/JNNN ''3/D+E+EEE$5HHHHHHHE$%CDDD* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6* 6X$#N333r3) additional_columnsrErFrHrIrJfeature_type_speclabel_type_specadditional_type_spec label_columnsrrz tf.TypeSpecrrztf.data.Datasetc  ddlmm}  ddl} n#t$rt dwxYwdt ddffd dtt tfddffd }d tt tj fdtt tt fd t| j tt | j ffdt| j tt | j ffffd    f d }  X}t|j||| | | | B @}t|j|| | %| jj| f}n#| jj| f}| j}| jjjj|j_||S)a Return a TF Dataset over this dataset. .. warning:: If your 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" ... ) >>> it = ds.iterator(); it DataIterator(Dataset(num_rows=?, schema=...)) If your model accepts a single tensor as input, specify a single feature column. >>> it.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. >>> it.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") >>> it = preprocessor.transform(ds).iterator() >>> it DataIterator(Concatenator +- Dataset(num_rows=?, schema=...)) >>> it.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 iterator. >>> import tensorflow as tf >>> it.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))) >>> it = ds.iterator() >>> it.to_tf(feature_columns="sepal length (cm)", label_columns="target", additional_columns="sample weights") <_OptionsDataset element_spec=(TensorSpec(shape=(None,), dtype=tf.float64, name='sepal length (cm)'), 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. >>> it.to_tf( ... feature_columns="sepal length (cm)", ... 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,), dtype=tf.float64, name='sepal length (cm)'), 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 will be 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 dataset is not divisible by the batch size, then the last batch will be smaller. Defaults to False. local_shuffle_buffer_size: If non-None, the data will be 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 will be 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 ``tf.data.Dataset`` that yields inputs and targets. r)convert_ndarray_to_tf_tensor get_type_specNztensorflow must be installed!columnr>c B|vrtd|d|dddS)NzYou specified 'z_' in `feature_columns`, `label_columns`, or `additional_columns`, but there's no column named 'z*' in the dataset. Valid column names are: r)r)r valid_columnss r1validate_columnz+DataIterator.to_tf..validate_columnsb]** @f@@%+@@0=@@@+*r3rclt|tr|D] }|dS|dSr5)rr)rrrs r1validate_columnsz,DataIterator.to_tf..validate_columnssY'4(( )%,,F#OF++++,, (((((r3rq type_speccxt|tr|Sfd|DS)NrcDi|]}|||S)rr;)rrrqrrs r1rzHDataIterator.to_tf..convert_batch_to_tensors..sN44&MYv->r3)rstr)rqrrrs` `r1convert_batch_to_tensorsz4DataIterator.to_tf..convert_batch_to_tensorssk '3'' Y33E'NiXXXX&  r3c3 K D]S}t|tsJ| }| }||fV>|}|||fVTdS)Nr)rr)rLrr)rqfeatureslabelsadditional_metadatarrrFrrHrrrrrIrJrEr0s r1 generatorz%DataIterator.to_tf..generators++!1%#*C#5 , @ @"%.....33?>O21=O&-"F******B*B 2"6+++' #F,??????/ @ @r3)r)output_signature)rrr tensorflow ImportErrorrrrr rnpndarrayTypeSpecTensorrwrnamesdataDatasetfrom_generatorOptions experimentalAutoShardPolicyOFFexperimental_distributeauto_shard_policy with_options)r0rrrrErFrHrIrJrrrrtfrrrwdatasetoptionsrrrrs```````````` @@@@r1to_tfzDataIterator.to_tfs3H         > # # # # # > > ><== = > C D       )eCI&6 )4 ) ) ) ) ) ) RZ( 3S >* R[$sBK/?*@@A  29d3 >22 3       @ @ @ @ @ @ @ @ @ @ @ @ @ @ @ @ @4  $(?[[]]F --M  _ - - -  ] + + + - fo N N N +mFMJJJO  ).B.J[[]]F --M  / 0 0 0#0=AS#T#T#T  )go44%#("5GGgo44->,P5G'//## G 0 4 '9##G,,,s9r%cddlm}|\}}}t|}t ||}t t||j}|||S)a7Execute and materialize this data iterator into object store memory. .. note:: This method triggers the execution and materializes all blocks of the iterator, returning its contents as a :class:`~ray.data.dataset.MaterializedDataset` for further processing. r)r%) input_data) ray.data.datasetr%r@rrryrr_context)r0r%rNrW_ ref_bundlesexecution_plan logical_plans r1 materializezDataIterator.materializes 988888%)%A%A%C%C"%+,, &ud.>.>.@.@AA"  - - -  #  #"     r3)r>r&)r>r%)*r8r9r:__doc__abcabstractmethodr rrr rboolr@r"intrrrrMrrQrrrLr`rrprWrwr!ryrr rrrrrr rrrr;r3r1r=r=Hs3$   x "H\$:D@ A      !"&/37,05 5 5 5  5 sm 5  5 $,C=5 %SM5  ) 5 5 5 Y5 n9 ( 39 9999!"&/37,0GK7;47474747 47 sm 47  47$,C=47%SM47h {N'BCD47xs 3447 ) 47474747l!#!!!!88DcN3888Y8<  s   Y        [    !"$'KO.BB C 4%]34 ( -m!4d3 ;M6NN O 4444$,C=4%SM4 !%!4"$(#4$ ,%4444B IM !37,0LPJN f-f-f-sDI~.f-S$s)^,f- "(3-$s)1D"DE f-  f-f-f-$,C=f-%SMf-!S-5G0H!HIf-}d3 3E.FFGf-$ ] #Xd3 3E.F%G G f- !f-f-f-Yf-P   Y   r3r=)?rr^rtypingrrrrrrr r r r r rrr.ray.data._internal.block_batching.iter_batchesr'ray.data._internal.execution.interfacesr%ray.data._internal.logical.interfacesr8ray.data._internal.logical.operators.input_data_operatorrray.data._internal.planrray.data._internal.statsrray.data.blockrrrray.data.collate_fnrrrrrrrr ray.data.contextr!ray.util.annotationsr"r#rrrrr$r%r&r'r(r)r*r,ABCr=DatasetIteratorr;r3r1r s                             HHHHHH============NNNNNN111111111111HHHHHHHHHH                    )(((((AAAAAAAA LLL GCLL # # # # #HQK # # # } } } } } 37} }  } Br3