&`i ,F,ddlZddlZddlZddlZddlmZmZmZmZm Z m Z m Z m Z m Z mZmZddlZddlmZddlZddlmZddlmZddlmZddlmZddlmZdd l m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z'dd l(m)Z)ddl*m+Z+ddl,m-Z-m.Z.ddl/m0Z0m1Z1m2Z2ddl3m4Z4m5Z5ddl6m7Z7ddl8m9Z9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBddlCmDZDddlEmFZFddlGmHZHddlImJZJddlKmLZLddlMmNZNddlOmPZPddlQmRZRdd lSmTZTdd!lUmVZVmWZWmXZXmYZYmZZZdd"l[m\Z\dd#l]m^Z^dd$l_m`Z`dd%lambZbdd&lcmdZdmeZemfZfmgZgmhZhdd'limjZjmkZkmlZldd(lmmnZndd)lompZpmqZqdd*lrmsZsmtZtmuZumvZvdd+lwmxZxdd,lymzZzm{Z{dd-l|m}Z}m~Z~dd.lmZdd/lmZdd0lmZmZdd1lmZer>ddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZdd2lmZdd3lmZdd4lGmZed5ZejeZed6e ejfd7Zed8dd9d:e ed;edZed8ddd?d@ed;edAe edjddd dReee efdSe dTd;edUe eeefdre esdVe evdWedXedYed[e ee d\ezfdZe e edAe edzfrom_blocks..s 555U#'%..555c6g|]}tj|SrZ)rA from_blockr^s rarbzfrom_blocks..s$VVVe/:5AAVVVrcr2Nmetadataparent) r2r7r9rB get_currentcopyr0_contextrD)rW block_refsmeta_with_schemafrom_blocks_opexecution_plan logical_plans ra from_blocksrqs65f555JVVvVVV ,<==N"|-=>tLLL!!&&((N~~/FGGL   rc parallelismoverride_num_blocksitemsrtrureturncddl}t||}|dkrtd|t|tjtj\}}}tt||}|dkr!tt||\}}nd\}}g}g} | |D]%} tj} t} | |zt| |z} | dz|zt| dz|z}| | |D]B}||}t!|t"jjsd|i}| |C| }|t j|| t1j|| 't5|| }t7t9d| id tj}t=||j}tA||S) aCreate a :class:`~ray.data.Dataset` from a list of local Python objects. Use this method to create small datasets from data that fits in memory. The column name defaults to "item". Examples: >>> import ray >>> ds = ray.data.from_items([1, 2, 3, 4, 5]) >>> ds MaterializedDataset(num_blocks=..., num_rows=5, schema={item: int64}) >>> ds.schema() Column Type ------ ---- item int64 Args: items: List of local Python objects. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` holding the items. rNz$parallelism must be -1 or > 0, got: )rritem)statsr3rf)!builtins_get_num_output_blocks ValueErrorr:r\utilget_current_placement_grouprBriminlendivmodranger@builderr/ isinstance collectionsabcMappingaddbuildappendr]rArer3r7r9rjr0rkrD)rvrtrur|detected_parallelism_ block_size remainderrWrmir{r block_start block_endjrzr` from_items_oprorps ra from_itemsrsrDOOO(6IJJKaM MMNNN!8 ,,..!!""!Q s5zz+?@@a &s5zz3G H H II $ I&(F68 ^^0 1 1  &(((***ns1i'8'88 Uj(3q1ui+@+@@  Y77  A8DdKO$;<< &~ KK      cgenn%%% # .uEKKMM J J J    f&677M"{,<=dKKK!!&&((N}n.EFFL   rcrt concurrencyrunrcLt|dd}t||||S)aCreates a :class:`~ray.data.Dataset` from a range of integers [0..n). This function allows for easy creation of synthetic datasets for testing or benchmarking :ref:`Ray Data `. The column name defaults to "id". Examples: >>> import ray >>> ds = ray.data.range(10000) >>> ds Dataset(num_rows=10000, schema={id: int64}) >>> ds.map(lambda row: {"id": row["id"] * 2}).take(4) [{'id': 0}, {'id': 2}, {'id': 4}, {'id': 6}] Args: n: The upper bound of the range of integers. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` producing the integers from the range 0 to n. .. seealso:: :meth:`~ray.data.range_tensor` Call this method for creating synthetic datasets of tensor data. arrowid)r block_format column_namer)r'read_datasource)rrtrru datasources rarrs<V!17MMMJ /    rc)ry)shapertrrurcht|ddt|}t||||S)aCreates a :class:`~ray.data.Dataset` tensors of the provided shape from range [0...n]. This function allows for easy creation of synthetic tensor datasets for testing or benchmarking :ref:`Ray Data `. The column name defaults to "data". Examples: >>> import ray >>> ds = ray.data.range_tensor(1000, shape=(2, 2)) >>> ds Dataset( num_rows=1000, schema={data: ArrowTensorTypeV2(shape=(2, 2), dtype=int64)} ) >>> ds.map_batches(lambda row: {"data": row["data"] * 2}).take(2) [{'data': array([[0, 0], [0, 0]])}, {'data': array([[2, 2], [2, 2]])}] Args: n: The upper bound of the range of tensor records. shape: The shape of each tensor in the dataset. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` producing the tensor data from range 0 to n. .. seealso:: :meth:`~ray.data.range` Call this method to create synthetic datasets of integer data. tensordata)rrr tensor_shaper)r'tupler)rrrtrrurs ra range_tensorr)sMf! (U5\\J /    rcrtnum_cpusnum_gpusmemoryray_remote_argsrrurrrrrc (t||}tj} |i}|js7t t jd|d<d|vr | j|d<t||||}t|| |} t j } t|| jtj| | \} } } | | }t!dd|Did}t#|| ||rt%|nd || }t'|tj}t+||j}t/|| S) aRead a stream from a custom :class:`~ray.data.Datasource`. Args: datasource: The :class:`~ray.data.Datasource` to read data from. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. read_args: Additional kwargs to pass to the :class:`~ray.data.Datasource` implementation. Returns: :class:`~ray.data.Dataset` that reads data from the :class:`~ray.data.Datasource`. NFsoftscheduling_strategy)placement_groupr6cg|] }|j SrZ)rg)r_ read_tasks rarbz#read_datasource..sJJJ)9-JJJrcrfr)rt num_outputsrr)planrp)r}rBrisupports_distributed_readsrRr\get_runtime_context get_node_idrr< _get_datasource_or_legacy_readerrrr:target_max_block_sizeget_read_tasksr9r6rr7rjr0rkrC)rrtrrrrrru read_argsctxdatasource_or_legacy_readercur_pgrequested_parallelismr read_tasksr{read_oprorps rarrgsN)6IJJK  ! # #C  0 1O  # % % 1 1 3 32 2 2 -. O33141H-.8 O#C ## X 1 1 3 3F"9 !!!# ###1a-;;>> import ray >>> path = "s3://anonymous@air-example-data-2/6G-audio-data-LibriSpeech-train-clean-100-flac/train-clean-100/5022/29411/5022-29411-0000.flac" >>> ds = ray.data.read_audio(path) >>> ds.schema() Column Type ------ ---- amplitude ArrowTensorTypeV2(shape=(1, 191760), dtype=float) sample_rate int64 Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. filesystem: The pyarrow filesystem implementation to read from. These filesystems are specified in the `pyarrow docs `_. Specify this parameter 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. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. include_paths: If ``True``, include the path to each image. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file/directory paths in ``paths`` that are not found. Defaults to False. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :meth:`~ray.remote` in the read tasks. Returns: A :class:`~ray.data.Dataset` containing audio amplitudes and associated metadata. ropen_stream_args meta_providerrrrrrrrrrrrru)rrLr)rrrrrrrrrrrurrrrrs ra read_audiorskZ! /133)!1#'   J '/   rc)rrrrrinclude_timestampsrrrrrurrrrrc zt|||t|||| ||| }t||| | || | S)a Creates a :class:`~ray.data.Dataset` from video files. Each row in the resulting dataset represents a video frame. The column names default to "frame", "frame_index" and "frame_timestamp". Examples: >>> import ray >>> path = "s3://anonymous@ray-example-data/basketball.mp4" >>> ds = ray.data.read_videos(path) >>> ds.schema() Column Type ------ ---- frame ArrowTensorTypeV2(shape=(720, 1280, 3), dtype=uint8) frame_index int64 Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. filesystem: The pyarrow filesystem implementation to read from. These filesystems are specified in the `pyarrow docs `_. Specify this parameter 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. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. include_paths: If ``True``, include the path to each image. File paths are stored in the ``'path'`` column. include_timestmaps: If ``True``, include the frame timestamps from the video as a ``'frame_timestamp'`` column. ignore_missing_paths: If True, ignores any file/directory paths in ``paths`` that are not found. Defaults to False. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. ray_remote_args: kwargs passed to :meth:`~ray.remote` in the read tasks. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. Returns: A :class:`~ray.data.Dataset` containing video frames from the video files. ) rrrrrrrrrrr)r-rLr)rrrrrrrrrrrrurrrrrs ra read_videosr3snX! /133)!1#-'   J '/   rc) pipelineschemartrrrrrruuridatabase collectionrrzpymongoarrow.api.Schemac Ttd|||||d| } t| ||||| | | S)aCreate a :class:`~ray.data.Dataset` from a MongoDB database. The data to read from is specified via the ``uri``, ``database`` and ``collection`` of the MongoDB. The dataset is created from the results of executing ``pipeline`` against the ``collection``. If ``pipeline`` is None, the entire ``collection`` is read. .. tip:: For more details about these MongoDB concepts, see the following: - URI: https://www.mongodb.com/docs/manual/reference/connection-string/ - Database and Collection: https://www.mongodb.com/docs/manual/core/databases-and-collections/ - Pipeline: https://www.mongodb.com/docs/manual/core/aggregation-pipeline/ To read the MongoDB in parallel, the execution of the pipeline is run on partitions of the collection, with a Ray read task to handle a partition. Partitions are created in an attempt to evenly distribute the documents into the specified number of partitions. The number of partitions is determined by ``parallelism`` which can be requested from this interface or automatically chosen if unspecified (see the ``parallelism`` arg below). Examples: >>> import ray >>> from pymongoarrow.api import Schema # doctest: +SKIP >>> ds = ray.data.read_mongo( # doctest: +SKIP ... uri="mongodb://username:password@mongodb0.example.com:27017/?authSource=admin", # noqa: E501 ... database="my_db", ... collection="my_collection", ... pipeline=[{"$match": {"col2": {"$gte": 0, "$lt": 100}}}, {"$sort": "sort_field"}], # noqa: E501 ... schema=Schema({"col1": pa.string(), "col2": pa.int64()}), ... override_num_blocks=10, ... ) Args: uri: The URI of the source MongoDB where the dataset is read from. For the URI format, see details in the `MongoDB docs `_. database: The name of the database hosted in the MongoDB. This database must exist otherwise ValueError is raised. collection: The name of the collection in the database. This collection must exist otherwise ValueError is raised. pipeline: A `MongoDB pipeline `_, which is executed on the given collection with results used to create Dataset. If None, the entire collection will be read. schema: The schema used to read the collection. If None, it'll be inferred from the results of pipeline. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. mongo_args: kwargs passed to `aggregate_arrow_all() `_ in pymongoarrow in producing Arrow-formatted results. Returns: :class:`~ray.data.Dataset` producing rows from the results of executing the pipeline on the specified MongoDB collection. Raises: ValueError: if ``database`` doesn't exist. ValueError: if ``collection`` doesn't exist. )rrrrrrrrrtrrrurZ)r$r)rrrrrrtrrrrrru mongo_argsrs ra read_mongorshv!    J '/    rc project_iddatasetqueryc Tt|||} t| ||||||| S)a{ Create a dataset from BigQuery. The data to read from is specified via the ``project_id``, ``dataset`` and/or ``query`` parameters. The dataset is created from the results of executing ``query`` if a query is provided. Otherwise, the entire ``dataset`` is read. For more information about BigQuery, see the following concepts: - Project id: `Creating and Managing Projects `_ - Dataset: `Datasets Intro `_ - Query: `Query Syntax `_ This method uses the BigQuery Storage Read API which reads in parallel, with a Ray read task to handle each stream. The number of streams is determined by ``parallelism`` which can be requested from this interface or automatically chosen if unspecified (see the ``parallelism`` arg below). .. warning:: The maximum query response size is 10GB. Examples: .. testcode:: :skipif: True import ray # Users will need to authenticate beforehand (https://cloud.google.com/sdk/gcloud/reference/auth/login) ds = ray.data.read_bigquery( project_id="my_project", query="SELECT * FROM `bigquery-public-data.samples.gsod` LIMIT 1000", ) Args: project_id: The name of the associated Google Cloud Project that hosts the dataset to read. For more information, see `Creating and Managing Projects `_. dataset: The name of the dataset hosted in BigQuery in the format of ``dataset_id.table_id``. Both the dataset_id and table_id must exist otherwise an exception will be raised. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: Dataset producing rows from the results of executing the query (or reading the entire dataset) on the specified BigQuery dataset. )rrrr)rr) rrrrtrrrrrrurs ra read_bigqueryrsIR$z7RWXXXJ '/    rchive)rcolumnsrtrrrrtensor_column_schemarrrrrrrrurr.rc t| t| d|vrtjdtdt |fi|}|dd}|dd}|dd}t|||||||| | | | | | }t|||||||| S) aCreates a :class:`~ray.data.Dataset` from parquet files. Examples: Read a file in remote storage. >>> 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 Read a directory in remote storage. >>> ds = ray.data.read_parquet("s3://anonymous@ray-example-data/iris-parquet/") Read multiple local files. >>> ray.data.read_parquet( ... ["local:///path/to/file1", "local:///path/to/file2"]) # doctest: +SKIP Specify a schema for the parquet file. >>> import pyarrow as pa >>> fields = [("sepal.length", pa.float32()), ... ("sepal.width", pa.float32()), ... ("petal.length", pa.float32()), ... ("petal.width", pa.float32()), ... ("variety", pa.string())] >>> ds = ray.data.read_parquet("s3://anonymous@ray-example-data/iris.parquet", ... schema=pa.schema(fields)) >>> ds.schema() Column Type ------ ---- sepal.length float sepal.width float petal.length float petal.width float variety string The Parquet reader also supports projection and filter pushdown, allowing column selection and row filtering to be pushed down to the file scan. .. testcode:: import pyarrow as pa # Create a Dataset by reading a Parquet file, pushing column selection and # row filtering down to the file scan. ds = ray.data.read_parquet( "s3://anonymous@ray-example-data/iris.parquet", columns=["sepal.length", "variety"], filter=pa.dataset.field("sepal.length") > 5.0, ) ds.show(2) .. testoutput:: {'sepal.length': 5.1, 'variety': 'Setosa'} {'sepal.length': 5.4, 'variety': 'Setosa'} For further arguments you can pass to PyArrow as a keyword argument, see the `PyArrow API reference `_. Args: paths: A single file path or directory, or a list of file paths. Multiple directories are not supported. filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `pyarrow docs `_. Specify this parameter 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. If ``None``, this function uses a system-chosen implementation. columns: A list of column names to read. Only the specified columns are read during the file scan. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. tensor_column_schema: A dict of column name to PyArrow dtype and shape mappings for converting a Parquet column containing serialized tensors (ndarrays) as their elements to PyArrow tensors. This function assumes that the tensors are serialized in the raw NumPy array format in C-contiguous order (e.g., via `arr.tobytes()`). meta_provider: A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases you do not need to set this parameter. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to HIVE partitioning. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. arrow_parquet_args: Other parquet read options to pass to PyArrow. For the full set of arguments, see the `PyArrow API `_ include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` producing records read from the specified parquet files. filterzThe `filter` argument is deprecated and will not supported in a future release. Use `dataset.filter(expr=expr)` instead to filter rows. stackleveldataset_kwargsN _block_udfr) rrto_batch_kwargsrrrrrrrrrr) '_emit_meta_provider_deprecation_warningrKwarningswarnDeprecationWarning_resolve_parquet_argspopr&r)rrrrtrrrrrrrrrrrrruarrow_parquet_argsrrrrs ra read_parquetr\s$l,M:::'"""%%%  F      /  (++,>> import ray >>> path = "s3://anonymous@ray-example-data/batoidea/JPEGImages/" >>> ds = ray.data.read_images(path) >>> ds.schema() Column Type ------ ---- image ArrowTensorTypeV2(shape=(32, 32, 3), dtype=uint8) If you need image file paths, set ``include_paths=True``. >>> ds = ray.data.read_images(path, include_paths=True) >>> ds.schema() Column Type ------ ---- image ArrowTensorTypeV2(shape=(32, 32, 3), dtype=uint8) path string >>> ds.take(1)[0]["path"] 'ray-example-data/batoidea/JPEGImages/1.jpeg' If your images are arranged like: .. code:: root/dog/xxx.png root/dog/xxy.png root/cat/123.png root/cat/nsdf3.png Then you can include the labels by specifying a :class:`~ray.data.datasource.partitioning.Partitioning`. >>> import ray >>> from ray.data.datasource.partitioning import Partitioning >>> root = "s3://anonymous@ray-example-data/image-datasets/dir-partitioned" >>> partitioning = Partitioning("dir", field_names=["class"], base_dir=root) >>> ds = ray.data.read_images(root, size=(224, 224), partitioning=partitioning) >>> ds.schema() Column Type ------ ---- image ArrowTensorTypeV2(shape=(224, 224, 3), dtype=uint8) class string Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. filesystem: The pyarrow filesystem implementation to read from. These filesystems are specified in the `pyarrow docs `_. Specify this parameter 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. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. meta_provider: [Deprecated] A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases, you do not need to set this. If ``None``, this function uses a system-chosen implementation. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. arrow_open_file_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. By default, this filters out any file paths whose file extension does not match ``*.png``, ``*.jpg``, ``*.jpeg``, ``*.tiff``, ``*.bmp``, or ``*.gif``. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. size: The desired height and width of loaded images. If unspecified, images retain their original shape. mode: A `Pillow mode `_ describing the desired type and depth of pixels. If unspecified, image modes are inferred by `Pillow `_. include_paths: If ``True``, include the path to each image. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file/directory paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` producing tensors that represent the images at the specified paths. For information on working with tensors, read the :ref:`tensor data guide `. Raises: ValueError: if ``size`` contains non-positive numbers. ValueError: if ``mode`` is unsupported. N) rrrrrrrrrrrr)rrrr)rrrtrrrrrrrrrrrrrrrrurs ra read_imagesr!sR,M:::133    ##-)!1'   J '/    rc)linesrrtrrrrrrrrrrrrrrurc t| |r5|||d}|D]\}}|rtd|d| t} t ||| | | | || | }|r=t jjj j }t|fd|i|}nt|fd|i|}t||||||||S) aWCreates a :class:`~ray.data.Dataset` from JSON and JSONL files. For JSON file, the whole file is read as one row. For JSONL file, each line of file is read as separate row. Examples: Read a JSON file in remote storage. >>> import ray >>> ds = ray.data.read_json("s3://anonymous@ray-example-data/log.json") >>> ds.schema() Column Type ------ ---- timestamp timestamp[...] size int64 Read a JSONL file in remote storage. >>> ds = ray.data.read_json("s3://anonymous@ray-example-data/train.jsonl", lines=True) >>> ds.schema() Column Type ------ ---- input Read multiple local files. >>> ray.data.read_json( # doctest: +SKIP ... ["local:///path/to/file1", "local:///path/to/file2"]) Read multiple directories. >>> ray.data.read_json( # doctest: +SKIP ... ["s3://bucket/path1", "s3://bucket/path2"]) By default, :meth:`~ray.data.read_json` parses `Hive-style partitions `_ from file paths. If your data adheres to a different partitioning scheme, set the ``partitioning`` parameter. >>> ds = ray.data.read_json("s3://anonymous@ray-example-data/year=2022/month=09/sales.json") >>> ds.take(1) [{'order_number': 10107, 'quantity': 30, 'year': '2022', 'month': '09'}] Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. lines: [Experimental] If ``True``, read files assuming line-delimited JSON. If set, will ignore the ``filesystem``, ``arrow_open_stream_args``, and ``arrow_json_args`` parameters. filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `PyArrow docs `_. Specify this parameter 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. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. meta_provider: [Deprecated] A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases, you do not need to set this. If ``None``, this function uses a system-chosen implementation. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. By default, this filters out any file paths whose file extension does not match "*.json" or "*.jsonl". partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. By default, this function parses `Hive-style partitions `_. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to ``FileShuffleConfig``, you can pass a random seed to shuffle the input files, e.g. ``FileShuffleConfig(seed=42)``. Defaults to not shuffle with ``None``. arrow_json_args: JSON read options to pass to `pyarrow.json.read_json `_. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` producing records read from the specified paths. )rrarrow_json_args`z&` is not supported when `lines=True`. Nrtarget_output_size_bytesrr)rrvr~rLdictr\rcontextrBrirrrr)rrrrtrrrrrrrrrrrrrrurincompatible_paramsparamvaluefile_based_datasource_kwargsrrs ra read_jsonr ssD,M::: T$&<.   05577 T TLE5 T !RU!R!R!RSSS T355 #'/#)!1#' $ $ $   H  ( 4 4 6 6 L !*   %= +  )   + +  '/    rc)rrtrrrrrrrrrrrrrruc t||t}t|||||| | | | | | }t||||||||S)aCreates a :class:`~ray.data.Dataset` from CSV files. Examples: Read a file in remote storage. >>> import ray >>> ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") >>> ds.schema() Column Type ------ ---- sepal length (cm) double sepal width (cm) double petal length (cm) double petal width (cm) double target int64 Read multiple local files. >>> ray.data.read_csv( # doctest: +SKIP ... ["local:///path/to/file1", "local:///path/to/file2"]) Read a directory from remote storage. >>> ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris-csv/") Read files that use a different delimiter. For more uses of ParseOptions see https://arrow.apache.org/docs/python/generated/pyarrow.csv.ParseOptions.html # noqa: #501 >>> from pyarrow import csv >>> parse_options = csv.ParseOptions(delimiter="\t") >>> ds = ray.data.read_csv( ... "s3://anonymous@ray-example-data/iris.tsv", ... parse_options=parse_options) >>> ds.schema() Column Type ------ ---- sepal.length double sepal.width double petal.length double petal.width double variety string Convert a date column with a custom format from a CSV file. For more uses of ConvertOptions see https://arrow.apache.org/docs/python/generated/pyarrow.csv.ConvertOptions.html # noqa: #501 >>> from pyarrow import csv >>> convert_options = csv.ConvertOptions( ... timestamp_parsers=["%m/%d/%Y"]) >>> ds = ray.data.read_csv( ... "s3://anonymous@ray-example-data/dow_jones.csv", ... convert_options=convert_options) By default, :meth:`~ray.data.read_csv` parses `Hive-style partitions `_ from file paths. If your data adheres to a different partitioning scheme, set the ``partitioning`` parameter. >>> ds = ray.data.read_csv("s3://anonymous@ray-example-data/year=2022/month=09/sales.csv") >>> ds.take(1) [{'order_number': 10107, 'quantity': 30, 'year': '2022', 'month': '09'}] By default, :meth:`~ray.data.read_csv` reads all files from file paths. If you want to filter files by file extensions, set the ``file_extensions`` parameter. Read only ``*.csv`` files from a directory. >>> ray.data.read_csv("s3://anonymous@ray-example-data/different-extensions/", ... file_extensions=["csv"]) Dataset(num_rows=?, schema=...) Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `pyarrow docs `_. Specify this parameter 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. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. meta_provider: [Deprecated] A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases, you do not need to set this. If ``None``, this function uses a system-chosen implementation. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. By default, no files are filtered. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. By default, this function parses `Hive-style partitions `_. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. arrow_csv_args: CSV read options to pass to `pyarrow.csv.open_csv `_ when opening CSV files. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` producing records read from the specified paths. N) arrow_csv_argsrrrrrrrrrr)rrLrr)rrrtrrrrrrrrrrrrrrur rs raread_csvr sl,M:::355  %/#)!1#'   J '/    rczutf-8T)encodingdrop_empty_linesrrtrrrrrrrrrrrrrrurrct| | t} t||||| | | | ||| | }t||||||||S)aeCreate a :class:`~ray.data.Dataset` from lines stored in text files. The column name default to "text". Examples: Read a file in remote storage. >>> import ray >>> ds = ray.data.read_text("s3://anonymous@ray-example-data/this.txt") >>> ds.schema() Column Type ------ ---- text string Read multiple local files. >>> ray.data.read_text( # doctest: +SKIP ... ["local:///path/to/file1", "local:///path/to/file2"]) Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. encoding: The encoding of the files (e.g., "utf-8" or "ascii"). filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `PyArrow docs `_. Specify this parameter 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. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks and in the subsequent text decoding map task. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. meta_provider: [Deprecated] A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases, you do not need to set this. If ``None``, this function uses a system-chosen implementation. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. By default, no files are filtered. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` producing lines of text read from the specified paths. N) rrrrrrrrrrrr)rrLr)r)rrrrrtrrrrrrrrrrrrrrurs ra read_textr5sB,M:::355  )/#)!1#'   J '/    rcc t||t}t||||| | | | | | }t||||||||S)aCreate a :class:`~ray.data.Dataset` from records stored in Avro files. Examples: Read an Avro file in remote storage or local storage. >>> import ray >>> ds = ray.data.read_avro("s3://anonymous@ray-example-data/mnist.avro") >>> ds.schema() Column Type ------ ---- features list label int64 dataType string >>> ray.data.read_avro( # doctest: +SKIP ... ["local:///path/to/file1", "local:///path/to/file2"]) Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `PyArrow docs `_. Specify this parameter 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. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks and in the subsequent text decoding map task. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. meta_provider: [Deprecated] A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases, you do not need to set this. If ``None``, this function uses a system-chosen implementation. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. By default, no files are filtered. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` holding records from the Avro files. Nrr)rrLrr)rrrtrrrrrrrrrrrrrrurs ra read_avrorsv,M:::355  /#)!1#'   J '/    rc) rrtrrrrrrrrrruc t||t}t|| ||||||| ||  }t||| | S)a Create an Arrow dataset from numpy files. The column name defaults to "data". Examples: Read a directory of files in remote storage. >>> import ray >>> ray.data.read_numpy("s3://bucket/path") # doctest: +SKIP Read multiple local files. >>> ray.data.read_numpy(["/path/to/file1", "/path/to/file2"]) # doctest: +SKIP Read multiple directories. >>> ray.data.read_numpy( # doctest: +SKIP ... ["s3://bucket/path1", "s3://bucket/path2"]) Args: paths: A single file/directory path or a list of file/directory paths. A list of paths can contain both files and directories. filesystem: The filesystem implementation to read from. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_stream `_. numpy_load_args: Other options to pass to np.load. meta_provider: File metadata provider. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. If ``None``, this function uses a system-chosen implementation. partition_filter: Path-based partition filter, if any. Can be used with a custom callback to read only selected partitions of a dataset. By default, this filters out any file paths whose file extension does not match "*.npy*". partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. if setting to ``FileShuffleConfig``, the random seed can be passed toshuffle the input files, i.e. ``FileShuffleConfig(seed = 42)``. Defaults to not shuffle with ``None``. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: Dataset holding Tensor records read from the specified paths. N) numpy_load_argsrrrrrrrrrr)rrLr%r)rrrtrrrrrrrrrrurrs ra read_numpyr-s~V,M:::355   '/#)!1#'   J /    rc)rrtrrrrrrrrr tf_schemarrrrutfx_read_optionsrzschema_pb2.SchemarrUc ddl}t|d}|rL|dkr4 ddl}d}n,#t$rd}t dYnwxYw|t}t|| |||| | | | || }t||||||||}|r|j r|r| sdd l m }||S|S) aCreate a :class:`~ray.data.Dataset` from TFRecord files that contain `tf.train.Example `_ messages. .. tip:: Using the ``tfx-bsl`` library is more performant when reading large datasets (for example, in production use cases). To use this implementation, you must first install ``tfx-bsl``: 1. `pip install tfx_bsl --no-dependencies` 2. Pass tfx_read_options to read_tfrecords, for example: `ds = read_tfrecords(path, ..., tfx_read_options=TFXReadOptions())` .. warning:: This function exclusively supports ``tf.train.Example`` messages. If a file contains a message that isn't of type ``tf.train.Example``, then this function fails. Examples: >>> import ray >>> ray.data.read_tfrecords("s3://anonymous@ray-example-data/iris.tfrecords") Dataset(num_rows=?, schema=...) We can also read compressed TFRecord files, which use one of the `compression types supported by Arrow `_: >>> ray.data.read_tfrecords( ... "s3://anonymous@ray-example-data/iris.tfrecords.gz", ... arrow_open_stream_args={"compression": "gzip"}, ... ) Dataset(num_rows=?, schema=...) Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `PyArrow docs `_. Specify this parameter 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. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. when opening input files to read. To read a compressed TFRecord file, pass the corresponding compression type (e.g., for ``GZIP`` or ``ZLIB``), use ``arrow_open_stream_args={'compression': 'gzip'}``). meta_provider: [Deprecated] A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases, you do not need to set this. If ``None``, this function uses a system-chosen implementation. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. tf_schema: Optional TensorFlow Schema which is used to explicitly set the schema of the underlying Dataset. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. tfx_read_options: Specifies read options when reading TFRecord files with TFX. When no options are provided, the default version without tfx-bsl will be used to read the tfrecords. Returns: A :class:`~ray.data.Dataset` that contains the example features. Raises: ValueError: If a file contains a message that isn't a ``tf.train.Example``. rNFarmTzPlease install tfx-bsl package with `pip install tfx_bsl --no-dependencies`. This can help speed up the reading of large TFRecord files.) rrrrrrrrrr)rtrrrrrru)_infer_schema_and_transform) platformr processortfx_bslModuleNotFoundErrorloggerwarningrLr*rauto_infer_schema2ray.data._internal.datasource.tfrecords_datasourcer)rrrtrrrrrrrrrrrrrrurrtfx_readrrdsrs raread_tfrecordsr&spdOOO+M:::H H..00E99  NNNHH"   #  NNO      355 # /#)1#')   J '/   B  /  . /  / /       +*2... Is8&A! A!)topics time_range message_typesinclude_metadatarrtrrrrrrrrrrrrrur'r(r)r*ct| t|| t} |dg}|Wt|trBt |dkrt d|t|d|d}t||||||| | | |||| }t||||| | || S) aCreate a :class:`~ray.data.Dataset` from MCAP (Message Capture) files. MCAP is a format commonly used in robotics and autonomous systems for storing ROS2 messages and other time-series data. This reader provides predicate pushdown optimization for efficient filtering by topics, time ranges, and message types. Examples: :noindex: Read all MCAP files in a directory. >>> import ray >>> ds = ray.data.read_mcap("s3://bucket/mcap-data/") # doctest: +SKIP >>> ds.schema() # doctest: +SKIP Read with filtering for specific topics and time range. >>> from ray.data.datasource import TimeRange # doctest: +SKIP >>> ds = ray.data.read_mcap( # doctest: +SKIP ... "s3://bucket/mcap-data/", # doctest: +SKIP ... topics={"/camera/image_raw", "/lidar/points"}, # doctest: +SKIP ... time_range=TimeRange(start_time=1000000000, end_time=5000000000), # doctest: +SKIP ... message_types={"sensor_msgs/Image", "sensor_msgs/PointCloud2"} # doctest: +SKIP ... ) # doctest: +SKIP Alternatively, use a tuple for time range (backwards compatible). >>> ds = ray.data.read_mcap( # doctest: +SKIP ... "s3://bucket/mcap-data/", # doctest: +SKIP ... topics={"/camera/image_raw", "/lidar/points"}, # doctest: +SKIP ... time_range=(1000000000, 5000000000), # doctest: +SKIP ... ) # doctest: +SKIP Read multiple local files with include_paths. >>> ray.data.read_mcap( # doctest: +SKIP ... ["local:///path/to/file1.mcap", "local:///path/to/file2.mcap"], # doctest: +SKIP ... include_paths=True # doctest: +SKIP ... ) # doctest: +SKIP Read with topic filtering and metadata inclusion. >>> ds = ray.data.read_mcap( # doctest: +SKIP ... "data.mcap", # doctest: +SKIP ... topics={"/camera/image_raw", "/lidar/points"}, # doctest: +SKIP ... include_metadata=True, # doctest: +SKIP ... include_paths=True # doctest: +SKIP ... ) # doctest: +SKIP Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. topics: Optional list or set of topic names to include. If specified, only messages from these topics will be read. time_range: Optional time range for filtering messages by timestamp. Can be either a tuple of (start_time, end_time) in nanoseconds (for backwards compatibility) or a TimeRange object. Both values must be non-negative and start_time < end_time. message_types: Optional list or set of message type names (schema names) to include. Only messages with matching schema names will be read. include_metadata: Whether to include MCAP metadata fields in the output. Defaults to True. When True, includes schema, channel, and message metadata. filesystem: The PyArrow filesystem implementation to read from. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. meta_provider: A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases you do not need to set this parameter. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. file_extensions: A list of file extensions to filter files by. Defaults to ``["mcap"]``. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` producing records read from the specified MCAP files. Nmcaprz:Time range must be a tuple of (start_time, end_time): got rry) start_timeend_time) r'r(r)r*rrrrrrrrr) rrKrLrrrr~r#r"r)rr'r(r)r*rrtrrrrrrrrrrrrrurs ra read_mcapr/Asp,M:::'"""355 !(*Z"?"? z??a      *Q-*Q-PPP  #)#)!1#'J '/    rc)rrtrrrdecoder fileselect filerenamesuffixes verbose_openrrrrru expand_jsonr0r1r2r3r4r5ct||t}t||||| | ||||| | | |}t||||S)a Create a :class:`~ray.data.Dataset` from `WebDataset `_ files. Args: paths: A single file/directory path or a list of file/directory paths. A list of paths can contain both files and directories. filesystem: The filesystem implementation to read from. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. arrow_open_stream_args: Key-word arguments passed to `pyarrow.fs.FileSystem.open_input_stream `_. To read a compressed TFRecord file, pass the corresponding compression type (e.g. for ``GZIP`` or ``ZLIB``, use ``arrow_open_stream_args={'compression': 'gzip'}``). meta_provider: File metadata provider. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. If ``None``, this function uses a system-chosen implementation. partition_filter: Path-based partition filter, if any. Can be used with a custom callback to read only selected partitions of a dataset. decoder: A function or list of functions to decode the data. fileselect: A callable or list of glob patterns to select files. filerename: A function or list of tuples to rename files prior to grouping. suffixes: A function or list of suffixes to select for creating samples. verbose_open: Whether to print the file names as they are opened. shuffle: If setting to "files", randomly shuffle input files order before read. if setting to ``FileShuffleConfig``, the random seed can be passed toshuffle the input files, i.e. ``FileShuffleConfig(seed = 42)``. Defaults to not shuffle with ``None``. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. expand_json: If ``True``, expand JSON objects into individual samples. Defaults to ``False``. Returns: A :class:`~ray.data.Dataset` that contains the example features. Raises: ValueError: If a file contains a message that isn't a `tf.train.Example`_. .. _tf.train.Example: https://www.tensorflow.org/api_docs/python/tf/train/Example N) r0r1r2r3r4rrrrrrrr5r)rrLr.r)rrrtrrrr0r1r2r3r4rrrrrur5rs raread_webdatasetr7sL,M:::355 % !/#)#'J /    rc)rrrtrrrrrrrrrrrrruc t| | t} t||||| | | | | | }t||||||||S)aCreate a :class:`~ray.data.Dataset` from binary files of arbitrary contents. Examples: Read a file in remote storage. >>> import ray >>> path = "s3://anonymous@ray-example-data/pdf-sample_0.pdf" >>> ds = ray.data.read_binary_files(path) >>> ds.schema() Column Type ------ ---- bytes binary Read multiple local files. >>> ray.data.read_binary_files( # doctest: +SKIP ... ["local:///path/to/file1", "local:///path/to/file2"]) Read a file with the filepaths included as a column in the dataset. >>> path = "s3://anonymous@ray-example-data/pdf-sample_0.pdf" >>> ds = ray.data.read_binary_files(path, include_paths=True) >>> ds.take(1)[0]["path"] 'ray-example-data/pdf-sample_0.pdf' Args: paths: A single file or directory, or a list of file or directory paths. A list of paths can contain both files and directories. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `PyArrow docs `_. Specify this parameter 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. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. arrow_open_stream_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_file `_. meta_provider: [Deprecated] A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases, you do not need to set this. If ``None``, this function uses a system-chosen implementation. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. By default, no files are filtered. By default, this does not filter out any files. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to ``None``. ignore_missing_paths: If True, ignores any file paths in ``paths`` that are not found. Defaults to False. shuffle: If setting to "files", randomly shuffle input files order before read. If setting to :class:`~ray.data.FileShuffleConfig`, you can pass a seed to shuffle the input files. Defaults to not shuffle with ``None``. file_extensions: A list of file extensions to filter files by. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` producing rows read from the specified paths. N) rrrrrrrrrr)rrLrr)rrrrtrrrrrrrrrrrrrurs raread_binary_filesr9I sF,M:::355 ! #/#)!1'   J '/    rcMD5) shard_keys shard_hash_fnrtrrrrrrusqlconnection_factoryr;r<c t||||} | r;| dkr5|td| | stdt| |||||| | S)agRead from a database that provides a `Python DB API2-compliant `_ connector. .. note:: Parallelism is supported by databases that support sharding. This means that the database needs to support all of the following operations: ``MOD``, ``ABS``, and ``CONCAT``. You can use ``shard_hash_fn`` to specify the hash function to use for sharding. The default is ``MD5``, but other common alternatives include ``hash``, ``unicode``, and ``SHA``. If the database does not support sharding, the read operation will be executed in a single task. Examples: For examples of reading from larger databases like MySQL and PostgreSQL, see :ref:`Reading from SQL Databases `. .. testcode:: import sqlite3 import ray # Create a simple database connection = sqlite3.connect("example.db") connection.execute("CREATE TABLE movie(title, year, score)") connection.execute( """ INSERT INTO movie VALUES ('Monty Python and the Holy Grail', 1975, 8.2), ("Monty Python Live at the Hollywood Bowl", 1982, 7.9), ("Monty Python's Life of Brian", 1979, 8.0), ("Rocky II", 1979, 7.3) """ ) connection.commit() connection.close() def create_connection(): return sqlite3.connect("example.db") # Get all movies ds = ray.data.read_sql("SELECT * FROM movie", create_connection) # Get movies after the year 1980 ds = ray.data.read_sql( "SELECT title, score FROM movie WHERE year >= 1980", create_connection ) # Get the number of movies per year ds = ray.data.read_sql( "SELECT year, COUNT(*) FROM movie GROUP BY year", create_connection ) .. testcode:: :hide: import os os.remove("example.db") Args: sql: The SQL query to execute. connection_factory: A function that takes no arguments and returns a Python DB API2 `Connection object `_. shard_keys: The keys to shard the data by. shard_hash_fn: The hash function string to use for sharding. Defaults to "MD5". For other databases, common alternatives include "hash" and "SHA". This is applied to the shard keys. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. This is used for sharding when shard_keys is provided. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`Dataset` containing the queried data. )r=r;r<r>ryNz8shard_keys must be provided when override_num_blocks > 1zHDatabase does not support sharding. Please set override_num_blocks to 1.r)r(r~supports_shardingr) r=r>r;r<rtrrrrrrurs raread_sqlrA sT #- J 2Q66  WXX X++,?@@ Z  '/    rc)r;rtrrrrrruconnection_parametersc n ddl  fd} tj|| |d|||||||  S)a7Read data from a Snowflake data set. Example: .. testcode:: :skipif: True import ray connection_parameters = dict( user=..., account="ABCDEFG-ABC12345", password=..., database="SNOWFLAKE_SAMPLE_DATA", schema="TPCDS_SF100TCL" ) ds = ray.data.read_snowflake("SELECT * FROM CUSTOMERS", connection_parameters) Args: sql: The SQL query to execute. 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. shard_keys: The keys to shard the data by. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. This is used for sharding when shard_keys is provided. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A ``Dataset`` containing the data from the Snowflake data set. rNc(jjdiS)NrZ) connectorconnect)rB snowflakesrasnowflake_connection_factoryz4read_snowflake..snowflake_connection_factory s *y"*CC-BCCCrchash) r>r;r<rtrrrrrru)snowflake.connectorr\rrA) r=rBr;rtrrrrrrurHrGs ` @raread_snowflakerKN sytDDDDDD 8   7'/    rc) tablercatalogrrtrrrrrru warehouse_idrLrMc ddlm} d} tjd}|st dtjd}|sddlm}|r| jj  }| d}nt d |sAdd lm }|d dd}|sAdd lm }|d dd}||t d|rd|}|t d| ||||||}t!|||||| | | S)a Read a Databricks unity catalog table or Databricks SQL execution result. Before calling this API, set the ``DATABRICKS_TOKEN`` environment variable to your Databricks warehouse access token. .. code-block:: console export DATABRICKS_TOKEN=... If you're not running your program on the Databricks runtime, also set the ``DATABRICKS_HOST`` environment variable. .. code-block:: console export DATABRICKS_HOST=adb-..azuredatabricks.net .. note:: This function is built on the `Databricks statement execution API `_. Examples: .. testcode:: :skipif: True import ray ds = ray.data.read_databricks_tables( warehouse_id='...', catalog='catalog_1', schema='db_1', query='select id from table_1 limit 750000', ) Args: warehouse_id: The ID of the Databricks warehouse. The query statement is executed on this warehouse. table: The name of UC table you want to read. If this argument is set, you can't set ``query`` argument, and the reader generates query of ``select * from {table_name}`` under the hood. query: The query you want to execute. If this argument is set, you can't set ``table_name`` argument. catalog: (Optional) The default catalog name used by the query. schema: (Optional) The default schema used by the query. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`Dataset` containing the queried data. r)DatabricksUCDatasourcectd} ddl}|}|||jddS#t$r|t $r|wxYw)NzNo dbutils module found.r user_globaldbutils) RuntimeErrorIPython get_ipythonns_table ImportErrorKeyError)no_dbutils_errorrUip_shells ra get_dbutilsz+read_databricks_tables..get_dbutils s'(BCC # NNN**,,H&&$]3I> > # # #" " # # #" " #s .AADATABRICKS_TOKENzXPlease set environment variable 'DATABRICKS_TOKEN' to databricks workspace access token.DATABRICKS_HOST)is_in_databricks_runtimebrowserHostNamezYou are not in databricks runtime, please set environment variable 'DATABRICKS_HOST' to databricks workspace URL(e.g. "adb-..azuredatabricks.net").)get_spark_sessionzSELECT CURRENT_CATALOG()zSELECT CURRENT_DATABASE()Nz5Only one of 'query' and 'table' arguments can be set.zselect * from z3One of 'query' and 'table' arguments should be set.)hosttokenrNrMrr)rrtrrrrrru)6ray.data._internal.datasource.databricks_uc_datasourcerPosenvirongetr~ray.util.spark.utilsr_notebook entry_point getDbutils getContexttagsrar=collectr)rNrLrrMrrtrrrrrrurPr\rcrbr_rrars raread_databricks_tablesro s_` # # # JNN- . .E   1   :>>+ , ,D  AAAAAA # # % %  &2==??HHJJUUWW 88::>>"34488::DDS  V::::::##%%))*DEEMMOOPQRSTU V::::::""$$(()DEEMMOOPQRSTU U.PQQQ )((( }NOOO'' ! J '/    rcsnapshot) query_typefilters hudi_optionsstorage_optionsrrrrrru table_urirqrrrsrtc Vt|||||} t| ||||| | S)a" Create a :class:`~ray.data.Dataset` from an `Apache Hudi table `_. Examples: >>> import ray >>> ds = ray.data.read_hudi( # doctest: +SKIP ... table_uri="/hudi/trips", ... query_type="snapshot", ... filters=[("city", "=", "san_francisco")], ... ) >>> ds = ray.data.read_hudi( # doctest: +SKIP ... table_uri="/hudi/trips", ... query_type="incremental", ... hudi_options={ ... "hoodie.read.file_group.start_timestamp": "20230101123456789", ... "hoodie.read.file_group.end_timestamp": "20230201123456789", ... }, ... ) Args: table_uri: The URI of the Hudi table to read from. Local file paths, S3, and GCS are supported. query_type: The Hudi query type to use. Supported values are ``snapshot`` and ``incremental``. filters: Optional list of filters to apply to the Hudi table when the ``query_type`` is ``snapshot``. Each filter is a tuple of the form ``(column_name, operator, value)``. The operator can be one of ``"="``, ``"!="``, ``"<"``, ``"<="``, ``">"``, ``">="``. Currently, only filters on partition columns will be effective. hudi_options: A dictionary of Hudi options to pass to the Hudi reader. storage_options: Extra options that make sense for a particular storage connection. This is used to store connection parameters like credentials, endpoint, etc. See more explanation `here `_. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` producing records read from the Hudi table. )rurqrrrsrtrrrrrrru)rr) rurqrrrsrtrrrrrrurs ra read_hudirx= sVF !' J '/   rcdfzdaft.DataFramect}|J|tdkrtd|S)aCreate a :class:`~ray.data.Dataset` from a `Daft DataFrame `_. .. warning:: This function only works with PyArrow 13 or lower. For more details, see https://github.com/ray-project/ray/issues/53278. Args: df: A Daft DataFrame Returns: A :class:`~ray.data.Dataset` holding rows read from the DataFrame. Nz14.0.0zw`from_daft` only works with PyArrow 13 or lower. For more details, see https://github.com/ray-project/ray/issues/53278.)r parse_versionrTto_ray_dataset)rypyarrow_versions ra from_daftr~ sY*++O  & & &-1111 ?        rczdask.dataframe.DataFramecddl}ddlm}|}|j|d|i}ddlfdt fd|D}|S)a:Create a :class:`~ray.data.Dataset` from a `Dask DataFrame `_. Args: df: A `Dask DataFrame`_. Returns: A :class:`~ray.data.MaterializedDataset` holding rows read from the DataFrame. rN) ray_dask_get schedulerct|jrtj|St|tjr|St dt |)Nz5Expected a Ray object ref or a Pandas DataFrame, got )r DataFramer\r]rOr~type)rypandass rato_refzfrom_dask..to_ref sb b&* + + 72;;  CM * * IRRRR rcc g|]>}tt|j?SrZ)nextiterdaskvalues)r_partrs rarbzfrom_dask.. sAQQQDT$)**,,--.. / /QQQrc)r ray.util.daskr to_delayedpersistrfrom_pandas_refs)ryrr partitionspersisted_partitionsr%rrs @@ra from_daskr sKKK******J'4<L|LLMMM QQQQ`_. Args: df: A `Mars DataFrame`_, which must be executed by Mars-on-Ray. Returns: A :class:`~ray.data.MaterializedDataset` holding rows read from the DataFrame. rN)mars.dataframe dataframer|)rymdr%s ra from_marsr s. ##B''B Ircz modin.pandas.dataframe.DataFramecJddlm}||d}t|}|S)a@Create a :class:`~ray.data.Dataset` from a `Modin DataFrame `_. Args: df: A `Modin DataFrame`_, which must be using the Ray backend. Returns: A :class:`~ray.data.MaterializedDataset` rows read from the DataFrame. r)unwrap_partitionsaxis)-modin.distributed.dataframe.pandas.partitionsrr)ryrpartsr%s ra from_modinr s>POOOOO  bq ) ) )E % B Ircdfszpandas.DataFramec^ddl}t||jr|g}|Ht|dkr||d}n|d}t j||}ddlmtj }|j rfd|D}td|DS)aCreate a :class:`~ray.data.Dataset` from a list of pandas dataframes. Examples: >>> import pandas as pd >>> import ray >>> df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}) >>> ray.data.from_pandas(df) MaterializedDataset(num_blocks=1, num_rows=3, schema={a: int64, b: int64}) Create a Ray Dataset from a list of Pandas DataFrames. >>> ray.data.from_pandas([df, df]) MaterializedDataset(num_blocks=2, num_rows=6, schema={a: int64, b: int64}) Args: dfs: A pandas dataframe or a list of pandas dataframes. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` holding data read from the dataframes. rNryr))_cast_ndarray_columns_to_tensor_extensioncJg|]}| SrZ)rj)r_ryrs rarbzfrom_pandas..* s-RRR88CCRRRrcc6g|]}tj|SrZr[)r_rys rarbzfrom_pandas.., s 777RSWR[[777rc) rrrrconcatnp array_split"ray.air.util.data_batch_conversionrrBrienable_tensor_extension_castingr)rrupdaryrrs @ra from_pandasr s:#r|$$e& s88a<<))Ca)((CCa&CnS"566%''G.SRRRRcRRR 773777 8 88rccN t|tjr|g}nst|tr?|D];}t|tjst dt |>> import pandas as pd >>> import ray >>> df_ref = ray.put(pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})) >>> ray.data.from_pandas_refs(df_ref) MaterializedDataset(num_blocks=1, num_rows=3, schema={a: int64, b: int64}) Create a Ray Dataset from a list of Pandas Dataframes references. >>> ray.data.from_pandas_refs([df_ref, df_ref]) MaterializedDataset(num_blocks=2, num_rows=6, schema={a: int64, b: int64}) Args: dfs: A Ray object reference to a pandas dataframe, or a list of Ray object references to pandas dataframes. Returns: :class:`~ray.data.Dataset` holding data read from the dataframes. 6Expected list of Ray object refs, got list containing 8Expected Ray object ref or list of Ray object refs, got c:g|]}|SrZremote)r_ryget_metadata_schemas rarbz$from_pandas_refs..Y s("P"P"Pb#6#=#=b#A#A"P"P"Prcr5Nrfr num_returnsc:g|]}|SrZr)r_ry df_to_blocks rarbz$from_pandas_refs..h s' 0 0 0b;  b ! ! 0 0 0rc)rr\rOlistr~rrBrienable_pandas_blockr8r;rgr7r9rjr0r5rkrDr>mapzip) rryrmetadata_schemarorpresrWrrs @@rarr/ s@4#s}%%  e C      Bb#-00  WTRTXXWW    QtBxx Q Q   %''G"  ./NOO'"P"P"P"PC"P"P"PQQ& <"A$ O O O  # % % * * , ,  # sO , ,n.E  #     ##;KKKK 0 0 0 0C 0 0 0C!$S 22FOgo..O"|_=dKKK!!&&((N6?++^-DL   rcndarrayscnt|tjr|g}td|DS)aCreates a :class:`~ray.data.Dataset` from a list of NumPy ndarrays. The column name defaults to "data". Examples: >>> import numpy as np >>> import ray >>> arr = np.array([1]) >>> ray.data.from_numpy(arr) MaterializedDataset(num_blocks=1, num_rows=1, schema={data: int64}) Create a Ray Dataset from a list of NumPy arrays. >>> ray.data.from_numpy([arr, arr]) MaterializedDataset(num_blocks=2, num_rows=2, schema={data: int64}) Args: ndarrays: A NumPy ndarray or a list of NumPy ndarrays. Returns: :class:`~ray.data.Dataset` holding data from the given ndarrays. c6g|]}tj|SrZr[)r_ndarrays rarbzfrom_numpy.. s"EEECGG,,EEErc)rrrfrom_numpy_refs)rs ra from_numpyrx s<0(BJ'': EEHEEE F FFrcct|tjr|g}nst|tr?|D];}t|tjst dt |>> import numpy as np >>> import ray >>> arr_ref = ray.put(np.array([1])) >>> ray.data.from_numpy_refs(arr_ref) MaterializedDataset(num_blocks=1, num_rows=1, schema={data: int64}) Create a Ray Dataset from a list of NumPy array references. >>> ray.data.from_numpy_refs([arr_ref, arr_ref]) MaterializedDataset(num_blocks=2, num_rows=2, schema={data: int64}) Args: ndarrays: A Ray object reference to a NumPy ndarray or a list of Ray object references to NumPy ndarrays. Returns: :class:`~ray.data.Dataset` holding data from the given ndarrays. rrrrc<g|]}|SrZr)r_rrndarray_to_block_remotes rarbz#from_numpy_refs.. s* P P PG " ) )'3 7 7 P P Prcr4Nrf)rr\rOrr~rrBrir8r=rrrgr7r9rjr0r4rkrD) rrrrWrrorprrs @@rarr s8(CM**  : Hd # #    Ggs}55  ;+/==;;   VtG}} V V     ! # #C./?QOOO P P P P Px P P PC!$S 22FOgo..O"{O|dkrt dt |dkr||n|d t }|dkr" fd||D}n||zdz |z}g}||D]K}||z}||krn=t|||z } | || Lt ||kr? dd} | | g|t |z z|}td|DS)aCreate a :class:`~ray.data.Dataset` from a list of PyArrow tables. Examples: >>> import pyarrow as pa >>> import ray >>> table = pa.table({"x": [1]}) >>> ray.data.from_arrow(table) MaterializedDataset(num_blocks=1, num_rows=1, schema={x: int64}) Create a Ray Dataset from a list of PyArrow tables. >>> ray.data.from_arrow([table, table]) MaterializedDataset(num_blocks=2, num_rows=2, schema={x: int64}) Args: tables: A PyArrow table, or a list of PyArrow tables, or its streaming format in bytes. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` holding data from the PyArrow tables. rNzoverride_num_blocks must be > 0ryc<g|]}ddS)r)slice)r_rcombined_tables rarbzfrom_arrow.. s6/0$$Q**rcc6g|]}tj|SrZr[)r_ts rarbzfrom_arrow.. s 7771CGAJJ777rc)r|pyarrowrTablebytesr~r concat_tablesrrrrextendfrom_arrow_refs) rrur|pa total_rows batch_sizeslicesrstartlength empty_tablers @ra from_arrowr s@OOO&28U+,,& ! # #>?? ?58[[1__))&111&QR)(( ??4CVVJF^^$788 C CJJ&&EZe);<< n225&AABBBB6{{000,221a88  {m/BS[[/PQRRRF 77777 8 88rcct|tjr|g}tttjfd|D}t td|idtj }tt|||j }t||S)a#Create a :class:`~ray.data.Dataset` from a list of Ray object references to PyArrow tables. Examples: >>> import pyarrow as pa >>> import ray >>> table_ref = ray.put(pa.table({"x": [1]})) >>> ray.data.from_arrow_refs(table_ref) MaterializedDataset(num_blocks=1, num_rows=1, schema={x: int64}) Create a Ray Dataset from a list of PyArrow table references >>> ray.data.from_arrow_refs([table_ref, table_ref]) MaterializedDataset(num_blocks=2, num_rows=2, schema={x: int64}) Args: tables: A Ray object reference to Arrow table, or list of Ray object references to Arrow tables, or its streaming format in bytes. Returns: :class:`~ray.data.Dataset` holding data read from the tables. c:g|]}|SrZr)r_rrs rarbz#from_arrow_refs..? s(MMM299!<<MMMrcr1Nrf)rr\rOr8r;rgr7r9rBrirjr0r1rkrD)rrrorprs @rarr s<&#-((*+JKKgMMMMfMMMNNO"{O#..". Example can be found at https://github.com/delta-io/delta-sharing/blob/main/README.md#quick-start limit: A non-negative integer. Load only the ``limit`` rows if the parameter is specified. Use this optional parameter to explore the shared table without loading the entire table into memory. version: A non-negative integer. Load the snapshot of the table at the specified version. timestamp: A timestamp to specify the version of the table to read. json_predicate_hints: Predicate hints to be applied to the table. For more details, see: https://github.com/delta-io/delta-sharing/blob/main/PROTOCOL.md#json-predicates-for-filtering. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. concurrency: The maximum number of Ray tasks to run concurrently. Set this to control the number of tasks to run concurrently. This doesn't change the total number of tasks run or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`Dataset` containing the queried data. Raises: ValueError: If the URL is not properly formatted or if there is an issue with the Delta Sharing table connection. )rrrrrrw)rr\rr) rrrrrrrrrrrurs raread_delta_sharing_tablesrN s^Z( 1 J 8 # #'/ $  rczpyspark.sql.DataFramec`ddl}t||}|j||S)aCreate a :class:`~ray.data.Dataset` from a `Spark DataFrame `_. Args: df: A `Spark DataFrame`_, which must be created by RayDP (Spark-on-Ray). parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.MaterializedDataset` holding rows read from the DataFrame. rN)raydpr}sparkspark_dataframe_to_ray_dataset)ryrtrurs ra from_sparkr s5*LLL(6IJJK ; 5 5b+ F FFrc)zdatasets.Datasetzdatasets.IterableDatasetc ddl}ddlm}ddlm}t ||j|jfrP ||}t|dkrddl }g} |D]} | | dd} | j dkr| | jn&td | j d | d h#|j$r+} td | d | d Yd} ~ d} ~ wwxYw| st%dddl} | jj}t/| ||||ddt$|giSn,#t$|f$rtdYnwxYwt ||jrt1|||||St ||jr0|d}t5|dd|}|St ||j|jfr=t;|}t?d|dd|dtAdtC|)aiRead a Hugging Face Dataset into a Ray Dataset. Creates a :class:`~ray.data.MaterializedDataset` from a `Hugging Face Datasets Dataset `_ or a :class:`~ray.data.Dataset` from a `Hugging Face Datasets IterableDataset `_. It is recommended to use :func:`~ray.data.read_parquet` with the ``HfFileSystem`` filesystem to read Hugging Face datasets rather than ``from_huggingface``. See :ref:`Loading Hugging Face datasets ` for more details. Args: dataset: A `Hugging Face Datasets Dataset`_ or `Hugging Face Datasets IterableDataset`_. `DatasetDict `_ and `IterableDatasetDict `_ are not supported. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` holding rows from the `Hugging Face Datasets Dataset`_. rN)ClientResponseError)HuggingFaceDatasourceT)allow_redirectstimeoutzUnexpected status z resolving z$ from Hugging Face Hub parquet fileszFailed to resolve z: zDNo resolvable Parquet URLs found from Hugging Face Hub parquet filesretry_exceptions)rtrrrurrz]Distributed read via Hugging Face Hub parquet files failed, falling back on single node read.rrrra9You provided a Hugging Face DatasetDict or IterableDatasetDict, which contains multiple datasets, but `from_huggingface` now only accepts a single Hugging Face Dataset. To convert just a single Hugging Face Dataset to a Ray Dataset, specify a split. For example, `ray.data.from_huggingface(my_dataset_dictionary['z'])`. Available splits are .z0`dataset` must be a `datasets.Dataset`, but got )"datasetsaiohttp.client_exceptionsr4ray.data._internal.datasource.huggingface_datasourcerrIterableDatasetrClist_parquet_urls_from_datasetrrequestshead status_coderrr r!RequestExceptionFileNotFoundErrorfsspec.implementations.httpimplementationshttpHTTPFileSystemrr with_formatr DatasetDictIterableDatasetDictrkeysr TypeErrorr)rrtrrurrr file_urlsr resolved_urlsrrespefsspecr hf_ds_arrowray_dsavailable_keyss rafrom_huggingfacer sHOOO======'H4h6FGHH87 .LLWUUI9~~!! " $  C '}}S$PQ}RR+s22)00::::"NN!BT5E!B!BRU!B!B!B$4______ %+^3222-2AACC#! +# +(;%)*->@S,T%    A"\"#67    NN4       '8344  ! !' 2 2 2## 3     '8+,, ))'22 KN@STTT 'H0(2NOPP gllnn--  6  "  6 6 %3  6 6 6    NtG}} N N   s=2E"A$CE C<!C72E7C<`_. This function is inefficient. Use it to read small datasets or prototype. .. warning:: If your dataset is large, this function may execute slowly or raise an out-of-memory error. To avoid issues, read the underyling data with a function like :meth:`~ray.data.read_images`. .. note:: This function isn't parallelized. It loads the entire dataset into the local node's memory before moving the data to the distributed object store. Examples: >>> import ray >>> import tensorflow_datasets as tfds >>> dataset, _ = tfds.load('cifar10', split=["train", "test"]) # doctest: +SKIP >>> ds = ray.data.from_tf(dataset) # doctest: +SKIP >>> ds # doctest: +SKIP MaterializedDataset( num_blocks=..., num_rows=50000, schema={ id: binary, image: ArrowTensorTypeV2(shape=(32, 32, 3), dtype=uint8), label: int64 } ) >>> ds.take(1) # doctest: +SKIP [{'id': b'train_16399', 'image': array([[[143, 96, 70], [141, 96, 72], [135, 93, 72], ..., [ 96, 37, 19], [105, 42, 18], [104, 38, 20]], ..., [[195, 161, 126], [187, 153, 123], [186, 151, 128], ..., [212, 177, 147], [219, 185, 155], [221, 187, 157]]], dtype=uint8), 'label': 7}] Args: dataset: A `TensorFlow Dataset`_. Returns: A :class:`MaterializedDataset` that contains the samples stored in the `TensorFlow Dataset`_. )rras_numpy_iteratorrs rafrom_tfrPs'r d7446677 8 88rcztorch.utils.data.Dataset local_readci}|r7ttjddd}t t ||dS)aCreate a :class:`~ray.data.Dataset` from a `Torch Dataset `_. The column name defaults to "data". .. note:: The input dataset can either be map-style or iterable-style, and can have arbitrarily large amount of data. The data will be sequentially streamed with one single read task. Examples: >>> import ray >>> from torchvision import datasets >>> dataset = datasets.MNIST("data", download=True) # doctest: +SKIP >>> ds = ray.data.from_torch(dataset) # doctest: +SKIP >>> ds # doctest: +SKIP MaterializedDataset(num_blocks=..., num_rows=60000, schema={item: object}) >>> ds.take(1) # doctest: +SKIP {"item": (, 5)} Args: dataset: A `Torch Dataset`_. local_read: If ``True``, perform the read as a local read. Returns: A :class:`~ray.data.Dataset` containing the Torch dataset samples. Frr)rrrry)rru)rRr\rrrr+)rrrs ra from_torchrs}DO  #A'))5577$$$    ((('    rc*) row_filterrtselected_fields snapshot_id scan_kwargscatalog_kwargsrrrrrutable_identifierrrSrrrrc ddlm} |tjdtd|dkrtjdtd| |||||| } t | ||| | | | }|S) a Create a :class:`~ray.data.Dataset` from an Iceberg table. The table to read from is specified using a fully qualified ``table_identifier``. Using PyIceberg, any intended row filters, selection of specific fields and picking of a particular snapshot ID are applied, and the files that satisfy the query are distributed across Ray read tasks. The number of output blocks is determined by ``override_num_blocks`` which can be requested from this interface or automatically chosen if unspecified. .. tip:: For more details on PyIceberg, see - URI: https://py.iceberg.apache.org/ Examples: >>> import ray >>> from ray.data.expressions import col #doctest: +SKIP >>> # Read the table and apply filters using Ray Data expressions >>> ds = ray.data.read_iceberg( #doctest: +SKIP ... table_identifier="db_name.table_name", ... catalog_kwargs={"name": "default", "type": "glue"} ... ).filter(col("column_name") == "literal_value") >>> # Select specific columns >>> ds = ds.select_columns(["col1", "col2"]) #doctest: +SKIP Args: table_identifier: Fully qualified table identifier (``db_name.table_name``) row_filter: **Deprecated**. Use ``.filter()`` method on the dataset instead. A PyIceberg :class:`~pyiceberg.expressions.BooleanExpression` to use to filter the data *prior* to reading. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. selected_fields: **Deprecated**. Use ``.select_columns()`` method on the dataset instead. Which columns from the data to read, passed directly to PyIceberg's load functions. Should be an tuple of string column names. snapshot_id: Optional snapshot ID for the Iceberg table, by default the latest snapshot is used scan_kwargs: Optional arguments to pass to PyIceberg's Table.scan() function (e.g., case_sensitive, limit, etc.) catalog_kwargs: Optional arguments to pass to PyIceberg's catalog.load_catalog() function (e.g., name, type, etc.). For the function definition, see `pyiceberg catalog `_. ray_remote_args: Optional arguments to pass to :func:`ray.remote` in the read tasks. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources, and capped at the number of physical files to be read. You shouldn't manually set this value in most cases. Returns: :class:`~ray.data.Dataset` with rows from the Iceberg table. r)IcebergDatasourceNzThe 'row_filter' parameter is deprecated and will be removed in a future release. Use the .filter() method on the dataset instead. For example: ds = ray.data.read_iceberg(...).filter(col('column') > 5)rrrzThe 'selected_fields' parameter is deprecated and will be removed in a future release. Use the .select_columns() method on the dataset instead. For example: ds = ray.data.read_iceberg(...).select_columns(['col1', 'col2']))rrrrrr)rrtrrrrur)0ray.data._internal.datasource.iceberg_datasourcer!rrrr)rrrtrrrrrrrrrur!rrs ra read_icebergr#sXSRRRRR  U      &    \      #")'% J/'G Nrc) rrrrtscanner_optionsrrrrrrurr$c Xt||||||} t| |||| | | S)aX Create a :class:`~ray.data.Dataset` from a `Lance Dataset `_. Examples: >>> import ray >>> ds = ray.data.read_lance( # doctest: +SKIP ... uri="./db_name.lance", ... columns=["image", "label"], ... filter="label = 2 AND text IS NOT NULL", ... ) Args: uri: The URI of the Lance dataset to read from. Local file paths, S3, and GCS are supported. version: Load a specific version of the Lance dataset. This can be an integer version number or a string tag. By default, the latest version is loaded. columns: The columns to read. By default, all columns are read. filter: Read returns only the rows matching the filter. By default, no filter is applied. storage_options: Extra options that make sense for a particular storage connection. This is used to store connection parameters like credentials, endpoint, etc. For more information, see `Object Store Configuration `_. scanner_options: Additional options to configure the `LanceDataset.scanner()` method, such as `batch_size`. For more information, see `LanceDB API doc `_ ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` producing records read from the Lance dataset. )rrrrrtr$rw)r!r) rrrrrtr$rrrrrrurs ra read_lancer&<sY~! '' J '/   rc) rrorder_byclient_settings client_kwargsrrrrrrudsnr'r(r)c Zt|||||||} t| ||| | | | S)aZ Create a :class:`~ray.data.Dataset` from a ClickHouse table or view. Examples: >>> import ray >>> ds = ray.data.read_clickhouse( # doctest: +SKIP ... table="default.table", ... dsn="clickhouse+http://username:password@host:8124/default", ... columns=["timestamp", "age", "status", "text", "label"], ... filter="age > 18 AND status = 'active'", ... order_by=(["timestamp"], False), ... ) Args: table: Fully qualified table or view identifier (e.g., "default.table_name"). dsn: A string in standard DSN (Data Source Name) HTTP format (e.g., "clickhouse+http://username:password@host:8124/default"). For more information, see `ClickHouse Connection String doc `_. columns: Optional list of columns to select from the data source. If no columns are specified, all columns will be selected by default. filter: Optional SQL filter string that will be used in the WHERE statement (e.g., "label = 2 AND text IS NOT NULL"). The filter string must be valid for use in a ClickHouse SQL WHERE clause. Please Note: Parallel reads are not currently supported when a filter is set. Specifying a filter forces the parallelism to 1 to ensure deterministic and consistent results. For more information, see `ClickHouse SQL WHERE Clause doc `_. order_by: Optional tuple containing a list of columns to order by and a boolean indicating whether the order should be descending (True for DESC, False for ASC). Please Note: order_by is required to support parallelism. If not provided, the data will be read in a single task. This is to ensure that the data is read in a consistent order across all tasks. client_settings: Optional ClickHouse server settings to be used with the session/every request. For more information, see `ClickHouse Client Settings `_. client_kwargs: Optional additional arguments to pass to the ClickHouse client. For more information, see `ClickHouse Core Settings `_. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. Returns: A :class:`~ray.data.Dataset` producing records read from the ClickHouse table or view. )rLr*rrr'r(r)rw)rr)rLr*rrr'r(r)rrrrrrurs raread_clickhouser,s\P& '#J '/   rc) data_formatregion reader_kwargsrcr-r.r/cTt||||||}|S)aLoads a Unity Catalog table or files into a Ray Dataset using Databricks Unity Catalog credential vending, with automatic short-lived cloud credential handoff for secure, parallel, distributed access from external engines. This function works by leveraging Unity Catalog's credential vending feature, which grants temporary, least-privilege credentials for the cloud storage location backing the requested table or data files. It authenticates via the Unity Catalog REST API (Unity Catalog credential vending for external system access, `Databricks Docs `_), ensuring that permissions are enforced at the Databricks principal (user, group, or service principal) making the request. The function supports reading data directly from AWS S3, Azure Data Lake, or GCP GCS in standard formats including Delta and Parquet. .. note:: This function is experimental and under active development. Examples: Read a Unity Catalog Delta table: >>> import ray >>> ds = ray.data.read_unity_catalog( # doctest: +SKIP ... table="main.sales.transactions", ... url="https://dbc-XXXXXXX-XXXX.cloud.databricks.com", ... token="dapi...", ... region="us-west-2" ... ) >>> ds.show(3) # doctest: +SKIP Args: table: Unity Catalog table path in format ``catalog.schema.table``. url: Databricks workspace URL (e.g., ``"https://dbc-XXXXXXX-XXXX.cloud.databricks.com"``). token: Databricks Personal Access Token with ``EXTERNAL USE SCHEMA`` permission. data_format: Data format (``"delta"`` or ``"parquet"``). If not specified, inferred from table metadata. region: AWS region for S3 access (e.g., ``"us-west-2"``). Required for AWS, not needed for Azure/GCP. reader_kwargs: Additional arguments passed to the underlying Ray Data reader. Returns: A :class:`~ray.data.Dataset` containing the data from Unity Catalog. )base_urlrctable_full_namer-r.r/)r,read)rLrrcr-r.r/rEs raread_unity_catalogr4s=\&# I >>  rc)rrrtrrrrrrrrrrrupathc Tddl}d} ||n*#t$rtd|d|d|d|d wxYwdd lm}t |t std ||| }t|f||||| | | | | ||d |S) a Creates a :class:`~ray.data.Dataset` from Delta Lake files. Examples: >>> import ray >>> ds = ray.data.read_delta("s3://bucket@path/to/delta-table/") # doctest: +SKIP Args: path: A single file path for a Delta Lake table. Multiple tables are not yet supported. version: The version of the Delta Lake table to read. If not specified, the latest version is read. filesystem: The PyArrow filesystem implementation to read from. These filesystems are specified in the `pyarrow docs `_. Specify this parameter 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. If ``None``, this function uses a system-chosen implementation. columns: A list of column names to read. Only the specified columns are read during the file scan. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :meth:`~ray.remote` in the read tasks. meta_provider: A :ref:`file metadata provider `. Custom metadata providers may be able to resolve file metadata more quickly and/or accurately. In most cases you do not need to set this parameter. partition_filter: A :class:`~ray.data.datasource.partitioning.PathPartitionFilter`. Use with a custom callback to read only selected partitions of a dataset. partitioning: A :class:`~ray.data.datasource.partitioning.Partitioning` object that describes how paths are organized. Defaults to HIVE partitioning. shuffle: If setting to "files", randomly shuffle input files order before read. Defaults to not shuffle with ``None``. include_paths: If ``True``, include the path to each file. File paths are stored in the ``'path'`` column. 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 or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. **arrow_parquet_args: Other parquet read options to pass to PyArrow. For the full set of arguments, see the `PyArrow API `_ Returns: :class:`~ray.data.Dataset` producing records read from the specified parquet files. rN deltalakez"`ray.data.read_delta` depends on 'z', but 'z)' couldn't be imported. You can install 'z' by running `pip install z`.) DeltaTablez1Only a single Delta Lake table path is supported.)r) rrrtrrrrrrrru) importlib import_modulerXr7r8rstrr~ file_urisr)r5rrrrtrrrrrrrrrrrurr9packager8rs ra read_deltar>%s?bG ((((     # # #' # #6= # # # # #    %$$$$$ dC NLMMM JtW - - - 7 7 9 9E   '#)!#/     s 'Aonceearliestlatesti') trigger start_offset end_offsetkafka_auth_configrrrrru timeout_msbootstrap_serversrBrCrDrErFc |dkrtd|tjt |||||| d|||| | S)a] Read data from Kafka topics. This function supports bounded reads from Kafka topics, reading messages between a start and end offset. Only the "once" trigger is supported for now, which performs a single bounded read. Currently we only have one read task for each partition. Examples: .. testcode:: :skipif: True import ray # Read from a single topic with offset range ds = ray.data.read_kafka( topics="my-topic", bootstrap_servers="localhost:9092", start_offset=0, end_offset=1000, ) Args: topics: Kafka topic name(s) to read from. Can be a single topic name or a list of topic names. bootstrap_servers: Kafka broker addresses. Can be a single string or a list of strings. trigger: Trigger mode for reading. Only "once" is supported, which performs a single bounded read. start_offset: Starting position for reading. Can be: - int: Offset number - str: "earliest" end_offset: Ending position for reading (exclusive). Can be: - int: Offset number - str: "latest" kafka_auth_config: Authentication configuration. See KafkaAuthConfig for details. num_cpus: The number of CPUs to reserve for each parallel read worker. num_gpus: The number of GPUs to reserve for each parallel read worker. memory: The heap memory in bytes to reserve for each parallel read worker. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. override_num_blocks: Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn't manually set this value in most cases. timeout_ms: Timeout in milliseconds for every read task to poll until reaching end_offset (default 10000ms). If the read task does not reach end_offset within the timeout, it will stop polling and return the messages it has read so far. Returns: A :class:`~ray.data.Dataset` containing Kafka messages with the following schema: - offset: int64 - Message offset within partition - key: binary - Message key as raw bytes - value: binary - Message value as raw bytes - topic: string - Topic name - partition: int32 - Partition ID - timestamp: int64 - Message timestamp in milliseconds - timestamp_type: int32 - 0=CreateTime, 1=LogAppendTime - headers: map - Message headers (keys as strings, values as bytes) Raises: ValueError: If invalid parameters are provided. ImportError: If kafka-python is not installed. r?z.Only trigger='once' is supported. Got trigger=)r'rGrCrDrErFrr)rtrrrrru)r~r\rrr ) r'rGrBrCrDrErrrrrurFs ra read_kafkarIs~`&U'UUVVV 8 # #/%!/!    '/ $  rcr%rkwargsctj||jr(tjdt |jdi|}n|}|S)a&Generates reader. Args: ds: Datasource to read from. ctx: Dataset config to use. kwargs: Additional kwargs to pass to the legacy reader if `Datasource.create_reader` is implemented. Returns: The datasource or a generated legacy reader. z`create_reader` has been deprecated in Ray 2.9. Instead of creating a `Reader`, implement `Datasource.get_read_tasks` and `Datasource.estimate_inmemory_data_size`.rZ)rB _set_currentshould_create_readerrrr create_reader)r%rrJrs rarrsh S!!!  )  8     '7b&6&@&@&@&@##&(# &&rcc R"|dddfd }||d<|S)Nrr`rrwc Fddlm}D]u\}\tfd||D}|||||||}v |}|S)Nr)ArrowTensorArraycbg|]+}tj|,S))bufferdtype)rr as_buffer)r_bufrTrs rarbz=_resolve_parquet_args.._block_udf..1sB 5NNNrc)ray.data.extensionsrQrvrcolumn set_column_ensure_integer_indexr)r`rQtensor_col_namenp_colrTrexisting_block_udfrs @@rarz)_resolve_parquet_args.._block_udf)s < < < < < <3G3M3M3O3O  /%9#(<<#@#@((//@@#$//HH "-**511Lrc)r`rrwr)r)rrrr]s` @rarr"sX'/33L$GG       0,6<( rccP|dkrtdn||}|S)NrrzpThe argument ``parallelism`` is deprecated in Ray 2.10. Please specify argument ``override_num_blocks`` instead.)r r!rss rar}r}EsBb 8      () rccB|tjdtdSdS)NzNThe `meta_provider` argument is deprecated and will be removed after May 2025.)rrr)rs rarrSs8         ! rc)NN)N)rrNN)F)rrN)rloggingrertypingrrrrrrr r r r r numpyrpackaging.versionrr{r\ray._private.arrow_utilsrray._private.auto_init_hookr$ray.air.util.tensor_extensions.utilsr.ray.data._internal.datasource.audio_datasourcer-ray.data._internal.datasource.avro_datasourcer1ray.data._internal.datasource.bigquery_datasourcer/ray.data._internal.datasource.binary_datasourcer3ray.data._internal.datasource.clickhouse_datasourcer,ray.data._internal.datasource.csv_datasourcer6ray.data._internal.datasource.delta_sharing_datasourcer-ray.data._internal.datasource.hudi_datasourcer.ray.data._internal.datasource.image_datasourcerr-ray.data._internal.datasource.json_datasourcerrr.ray.data._internal.datasource.kafka_datasourcerr .ray.data._internal.datasource.lance_datasourcer!-ray.data._internal.datasource.mcap_datasourcer"r#.ray.data._internal.datasource.mongo_datasourcer$.ray.data._internal.datasource.numpy_datasourcer%0ray.data._internal.datasource.parquet_datasourcer&.ray.data._internal.datasource.range_datasourcer',ray.data._internal.datasource.sql_datasourcer(-ray.data._internal.datasource.text_datasourcer)r#r*.ray.data._internal.datasource.torch_datasourcer++ray.data._internal.datasource.uc_datasourcer,.ray.data._internal.datasource.video_datasourcer-3ray.data._internal.datasource.webdataset_datasourcer.+ray.data._internal.delegating_block_builderr/%ray.data._internal.logical.interfacesr03ray.data._internal.logical.operators.from_operatorsr1r2r3r4r52ray.data._internal.logical.operators.read_operatorr6ray.data._internal.planr7ray.data._internal.remote_fnr8ray.data._internal.statsr9ray.data._internal.utilr:r;r<r=r>ray.data.blockr?r@rAray.data.contextrBray.data.datasetrCrDray.data.datasourcerErFrGrHray.data.datasource.datasourcerI)ray.data.datasource.file_based_datasourcerJrK&ray.data.datasource.file_meta_providerrLrM ray.data.datasource.partitioningrN ray.typesrOray.util.annotationsrPrQray.util.scheduling_strategiesrRdaftrrmarsmodinrrpymongoarrow.api pymongoarrowpyspark tensorflowtftorchpyiceberg.expressionsrStensorflow_metadata.proto.v0rTrUrV getLogger__name__r rqintrrrfloatr;r_FILE_EXTENSIONSboolrrrrrTrrr r rrrr&r/callablerr7r9rArKrorxr~rrrrrrrrrrrrrrrrr#r&r,rr4r>rIrrr}rrZrcrars1                           444444 888888666666PPPPPPJJJJJJHHHHHHPPPPPPLLLLLLTTTTTTFFFFFFIHHHHH KJJJJJSSSSSSSSJJJJJJJJJJJJNNNNNNJJJJJJFFFFFFHHHHHHQQQQQQJJJJJJMMMMMMJJJJJJTTTTTTNNNNNN======DCCCCC111111999999111111 )(((((99999999 211111:9999988888888IIIIIIRKKKKKKOOOKKKLLLMMMNNNNNNLLL777777777777QQQQQQ GCLL  8 $ $U 8 )- QQQ 9QQ"# Q  QQQ Qh !%)- 000 00# 0 "# 0  000 0f !%)- ::: : : : # : "# : ::: :z  $ $"&*!%)-cccccuo c uo c UO c#s(^c#c"#c ccc cL W597;6:+/!&+:+K-1!%)- $ $"04!``` d3i `01`%T#s(^4 ` 23 ` <( ```d3i(`77#T) *`#`"#`uo`uo` UO` d38n-!````F W597;6:+/$!&+:+K-1!%)- $ $"04#``` d3i `01`%T#s(^4 ` 23 ` <( ````d3i(`77#T) *`#`"#`uo`uo` UO!`"d38n-#````F W &*26 $ $"&*!%)-kkk kkk tDz" k . / kkuokuok UOk#s(^k#k"#k kkkk\ W"R  $ $"&*!%)-RRRR c]R C=R  R uo RuoR UOR#s(^R#R"#R RRRRj 59#' $ $"&*RV486:+7<+?+?DH+<+M!%)-%AAA d3i A01Ad3i A  A uo AuoA UOA#s(^A#4U28U38_3L-M(M#NOA01A23A<(AeGG,.??@ AAA d3i(!A"##A$"#%A( )AAA AH V59 $ $"8<&*596:!%&*!&DH+:+K!%)-)ddd d3i d01d d uo d uo d UOd45d#s(^d#4S>2d23dd 5c? #d 3-dd !d"eGG,.??@ A#d$d3i(%d&#'d("#)d* +ddddN 48 $ $"&*7;8<6:!-f!5!5!&DH+?!%)-'ttt d3i t t01 t  t uo tuot UOt#s(^t%T#s(^4t45t23tttt eGG,.??@ A!t"d3i(#t$#%t&"#'t* +ttt tn 59 $ $"&*7;8<6:!-f!5!5!&DH+/!%)-%ppp d3i p01p p uo p uo p UOp#s(^p%T#s(^4p45p23ppppeGG,.??@ Ap d3i(!p"##p$"#%p( )ppp pf !48 $ $"047;8<6:!%!&DH+/!%)-)||| d3i || | 01 |  |uo|uo| UO|d38n-|%T#s(^4|45|23||| !|"eGG,.??@ A#|$d3i(%|&#'|("#)|* +||| |~ 59 $ $"047;8<6:!%!&DH+/!%)-%ttt d3i t01t t uo t uo t UOtd38n-t%T#s(^4t45t23tttteGG,.??@ At d3i(!t"##t$"#%t& 'ttt tn 597;8<6:!%!&DH+:+K!%)-aaa d3i a01a a %T#s(^4 a 45 a23aaaaeGG,.??@ Aad3i(a#a"#a  !aaa aH W59 $ $"&*7;8<6:!&/3DH+/!%)-37'kkk d3i k01k k uo k uo k UOk#s(^k%T#s(^4k45k23kkk+,keGG,.??@ Ak d3i(!k"##k$"#%k&/0'k( )kkkk\ W48>B:>!48 $ $"048<6:!%!&DH+/!%)-+aaa d3i a U49c#h./ 0auS#X 9:; a E$s)SX"567 a  a01aauoauoa UOad38n-a45a23aa !a"#a$eGG,.??@ A%a&d3i('a(#)a*"#+a, -aaaaH W597;8<6::>262604DH+/!%)-%___ d3i _01_ _ %T#s(^4 _ 45 _23_eD#x56 7_tX~./_tX~./_uT8^,-__eGG,.??@ A__d3i(_ #!_""##_$%_& '____D  48 $ $"&*7;8<6:!%!&DH+/!%)-%||| d3i ||01 |  | uo |uo| UO|#s(^|%T#s(^4|45|23|||eGG,.??@ A| d3i(!|"##|$"#%|& '||| |~ W '+ $ $"04!%)-AAA A Z0Ac# A  A  AuoAuoA UOAd38n-A#A"#A AAAAH W '+ $ $"&*!%)-JJJ JS>Jc# J  J uo JuoJ UOJ#s(^J#J"#J JJJJZ W !  $ $"04!%)-]]]] C=] C= ] c] ] SM ]]uo]uo] UO]d38n-]#]"#] ]]]]@ W!48-104 $ $"04!%)-RRRRRd5c3/0 1 R 4S>* R d38n- RuoRuoR UORd38n-R#R"#R RRRRj "w <  , 1D     F  , 1D       5:M " *.2929 !4(:#;; <29!#29292929 29jE y+,d9=O3P.QQ REEEEEP Grz4 +;;<GATGGG G:<Ibj)4 "*0E+FFG<<<<<~ *.C9C9C9 /5$u_e5K/L*MM NC9"#C9 C9C9C9 C9L- %./0 Yu_e34 56 8 -  ----` W !#*.04 $ $"!%)-]]] ] C=]c] ] } ] #3- ]d38n-]uo]uo] UO]#]"#] ]]]]@ "&)- GGGG#G"# G  GGG G4 !%)- B B A BB B #B "# B   '( B B B  B J 89 89898989 89v 33 '33 333 3l 37'-!%,0/304 $ $")-uuuuc../u u 38_ u # u$sCx.)uT#s(^,ud38n-uuouuou UOu"#u uuu up *.#' 040404 $ $"!%)-OOO OeCHo &Od3i O SM O d38n- Od38n-Od38n-OuoOuoO UOO#O"#O OOO Od W $( 1504.204 $ $"!%)-YYY Y Yd3i Y SM Y uT#Y_-. Yd38n-YDcN+Yd38n-YuoYuoY UOY#Y"#Y YYYYx W "& $(555 5 5 5 # 5 SM 5D>5 5555p W"t59#' $ $"04486:+7<+?+?-1!%)-#ttt T#Y t c]t01 t d3i t  tuotuot UOtd38n-t01t23t<(t77#T) *tt #!t""##ttttn W &4>0837 $ $"04)-aaa #tCy. !aS$s)^,aV_ a WZ001 a c78,,- a 0auoauoa UOad38n-a"#aa aaaaH'' ' ' :v  ''''BSW  "4U28U38_3L-M(M#NO  #s(^    H)-   !#       45        rc