o bi\&@sbddlZddlZddlZddlZddlmZmZmZmZm Z m Z m Z m Z m Z mZddlZddlmZddlZddlmZddlmZddlmZddlmZddlmZdd lm Z dd l!m"Z"dd l#m$Z$dd l%m&Z&dd l'm(Z(ddl)m*Z*ddl+m,Z,ddl-m.Z.m/Z/ddl0m1Z1m2Z2m3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9ddl:m;Z;ddlm?Z?ddl@mAZAddlBmCZCddlDmEZEddlFmGZGddlHmIZIddlJmKZKddlLmMZMddlNmOZOdd lPmQZQdd!lRmSZSmTZTmUZUmVZVmWZWdd"lXmYZYdd#lZm[Z[dd$l\m]Z]dd%l^m_Z_dd&l`maZambZbmcZcmdZddd'lemfZfmgZgmhZhdd(limjZjdd)lkmlZlmmZmdd*lnmoZompZpmqZqmrZrdd+lsmtZtdd,lumvZvmwZwdd-lxmyZymzZzdd.l{m|Z|dd/l}m~Z~dd0lmZdd1lmZmZmZdd2lmZerddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZdd3lmZdd4lmZdd5lDmZe d6ZeeZed7e effd8d9Zed:dd;de ed?emfd@dAZed:dddBdCed=edDe ed>e ed?elf dEdFZedGd:dddHdCedIe d=edDe ed>e ed?elf dJdKZeed:ddddLdMeqd=edNeeefdDe ed>e ed?elf dOdPZedQdRdddddSdSejdddddT dUeee efdVe dWdXe eeefdYe erdZe e~d[ed\ed]e e ed^ee d_dfdDe ed>e edNe eeeffd`daZedQdRdddddSdSdSeKjdddddb dUeee efdVe dWdXe eeefdYe erdZe e~d[edced\ed]e e ed^ee d_dfdDe ed>e edNe eeeffdddeZedQdRddd:ddddfdgedhediedje e edke dld=edNeeefdDe ed>e ed?elfdmdnZedQdR  d d:ddddLdoedpe edqe ed=edNeeefdDe ed>e ed?elfdrdsZeddd:dddde~dtddSddddu dUeee efdVe dWdve e ed=edNeeefdwe eee eje edxfffdye e|dYe erdZe e~d^e ee d_evfd[ed]e e edDe ed>e ed?elfdzd{Zed|dRdd:ddddddddSdSde.jddd}dUeee efdVe dWd=edye eodNeeefd~e eeefdYe erdZe~de e eefde ed[ed\ed^e ee d_evfd]e e edDe ed>e ed?elf"ddZeddd:dddddddSe;jddd dUeee efdVe dWdve e ed=edNeeefd~e eeefdwe eee eje edxfffdye eodYe erd^e ee d_evfd[ed]e e edDe ed>e ed?elfddZedSdd:dddde~dtdSdSde1ddddUeee efdedVe dWd=edNeeefdXe eeefdye eodYe erdZe~d[ed\ed^e ee d_evfd]e e edDe ed>e ed?elf ddZedd:dddde~dtdSdSddddd dUeee efdVe dWd=edNeeefdXe eeefdye eodYe erdZe~d[ed\ed^e ee d_evfd]e e edDe ed>e ed?elfddZedddd:ddddddSdSddddddUeee efdededVe dWd=edNe eeefdXe eeefdye eodYe erdZe~d[ed\ed^e ee d_evfd]e e edDe ed>e ed?elf"ddZedd:ddddddSdSddddd dUeee efdVe dWd=edNe eeefdXe eeefdye eodYe erdZe~d[ed\ed^e ee d_evfd]e e edDe ed>e ed?elfddZedd:dddddSdSde9jddd dUeee efdVe dWd=edXe eeefdye eodYe erdZe~d[ed\ed^e ee d_evfd]e e edDe ed>e ed?elfddZedQdRdd:dddddSdSddddddddUeee efdVe dWd=edNeeefdXe eeefdye eodYe erd[ed\ede dd^e ee d_evfd]e e edDe ed>e ede dd?elf ddZedQdRdd:ddddddddSddSddddSddUeee efdVe dWd=edXe eeefdye eodYe erde eeeeefde eeefde eeefde eeefded^e ee d_evfd[ed]e e edDe ed>e eded?elf$ddZedSdd:ddddddSddddd dUeee efd[edVe dWd=edNeeefdXe eeefdye eodYe erdZe~d\ed^e ee d_evfd]e e edDe ed>e ed?elfddZedQdRddd:dddddedegepfde eeded=edNe eeefdDe ed>e ed?elfddZedQdRddddd:dddddede edqe ede edke ed=edNe eeefdDe ed>e ed?elfddZedQdRddddddede eeefdNe eeefdDe ed>e ed?elf ddZeddd?elfdd„Zeddd?emfddńZeddd?emfddȄZeddd?emfdd˄Ze d!dede dfd>e ed?emfddτZedeede edfd?emfddфZedeeje ejfd?emfddԄZedeeeje eejfd?emfddքZeddלdedee edeffd>e ed?emfddۄZedeeedefe eedeffd?emfdd݄ZedQdRddddddddޜdede ede ede ede edNe eeefdDe ed>e ed?elfddZeddd;ddd=e ed>e ed?emfddZe :  d"dpedd=edDe ed>e ed?eemelff ddZedpdd?emfddZe Sd#dpdded?elfddZedd:ddddddddedeedfd=ede edxfde ede eeefde eeefdNe eeefd>e ed?elfddZedddddddddgedve e ede ede eeefde eeefdNe eeefdDe ed>e ed?elfddZedQdRddddddddddededve e ede ede e e eefde eeefde eeefdNe eeefdDe ed>e ed?elfdd ZedQdRddd deded ed e ed e ede ed?elfddZedQdRddd:ddde~dtddSddd deee efdVe dWdve e ed=edNe eeefdye e|dYe erdZe e~d^ee d_dfd[edDe ed>e efddZѐdeqdejded?eeqetffddZ d!dwe eee eje edxfffd?eeeffddZ : d$d=ed>e ed?efddZdye eod?dfddZdS(%N) TYPE_CHECKINGAnyCallableDictListLiteralOptionalTupleTypeVarUnion)parse)get_pyarrow_version)wrap_auto_init)_create_possibly_ragged_ndarray)AudioDatasource)AvroDatasource)BigQueryDatasource)BinaryDatasource)ClickHouseDatasource) CSVDatasource)DeltaSharingDatasource)HudiDatasource)IcebergDatasource)ImageDatasourceImageFileMetadataProvider)JSON_FILE_EXTENSIONSArrowJSONDatasourcePandasJSONDatasource)LanceDatasource)MongoDatasource)NumpyDatasource)ParquetBulkDatasource)ParquetDatasource)RangeDatasource) SQLDatasource)TextDatasource)TFRecordDatasource)TorchDatasource)UnityCatalogConnector)VideoDatasource)WebDatasetDatasource)DelegatingBlockBuilder) LogicalPlan) FromArrow FromBlocks FromItems FromNumpy FromPandas)Read) ExecutionPlan)cached_remote_fn) DatasetStats)_autodetect_parallelismget_table_block_metadata_schemandarray_to_blockpandas_df_to_arrow_block)BlockBlockExecStatsBlockMetadataWithSchema) DataContext)DatasetMaterializedDataset)BaseFileMetadataProvider Connection DatasourcePathPartitionFilter)Reader)FileShuffleConfig_validate_shuffle_arg)DefaultFileMetadataProviderFastFileMetadataProvider)ParquetMetadataProvider) Partitioning) ObjectRef) Deprecated DeveloperAPI PublicAPI)NodeAffinitySchedulingStrategy)BooleanExpression) schema_pb2)TFXReadOptionsTblockscCsZdd|D}dd|D}t||}ttd|iddt}t||j}t||S)aCreate a :class:`~ray.data.Dataset` from a list of blocks. This method is primarily used for testing. Unlike other methods like :func:`~ray.data.from_pandas` and :func:`~ray.data.from_arrow`, this method gaurentees that it won't modify the number of blocks. Args: blocks: List of blocks to create the dataset from. Returns: A :class:`~ray.data.Dataset` holding the blocks. cSg|]}t|qSrayput.0blockrVrVE/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/read_api.py zfrom_blocks..cSrUrV)r< from_blockrZrVrVr]r^r_r.Nmetadataparent) r.r3r5r= get_currentcopyr,_contextr?)rT block_refsmeta_with_schemafrom_blocks_opexecution_plan logical_planrVrVr] from_blockss   rl parallelismoverride_num_blocksitemsrorpreturncCspddl}t||}|dkrtd|t|tjt\}}}t t ||}|dkr7t t ||\}}nd\}}g}g} | |D]T} t } t} | |t | |} | d|t | d|}| | |D]}||}t|tjjszd|i}| |qi| }|t|| tj|| dqDt|| }ttd| idd t}t||j}t ||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)statsr/ra)!builtins_get_num_output_blocks ValueErrorr6rXutilget_current_placement_groupr=rdminlendivmodranger;builderr+ isinstance collectionsabcMappingaddbuildappendrYr<r`r/r3r5rer,rfr?)rqrorprvdetected_parallelism_ block_size remainderrTrhirur block_start block_endjrtr\ from_items_oprjrkrVrVr] from_itemssP"      rro concurrencyrpnrcCst|ddd}t||||dS)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)rrorrp datasourcerVrVr]r~s+r~)rs)shaperorrprcCs$t|ddt|d}t||||dS)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: numpy.ndarray(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)rrrorrprrVrVr] range_tensor%s0 rroray_remote_argsrrprrcKst||}t}|duri}|jsttdd|d<d|vr'|j|d<t |||}tj }t ||j t||d\} } } || } tddd| Didd } t|||| | r^t| nd ||}t| t}t||j}t||d 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. 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_groupr2cSsg|]}|jqSrV)rb)r[ read_taskrVrVr]r^sz#read_datasource..rar)planrk)rwr=rdsupports_distributed_readsrOrXget_runtime_context get_node_idr _get_datasource_or_legacy_readerryrzr6target_max_block_sizeget_read_tasksr5r2r|r3rer,rfr>)rrorrrp read_argsctxdatasource_or_legacy_readercur_pgrequested_parallelismr inmemory_size read_tasksruread_oprjrkrVrVr]r`s^         ralpha) stabilityF) filesystemarrow_open_stream_argspartition_filter partitioning include_pathsignore_missing_pathsfile_extensionsshufflerrprpathsrzpyarrow.fs.FileSystemrrrrrrrfilesc Cs.t|||t||||||d } t| | | | dS)a Creates a :class:`~ray.data.Dataset` from audio files. The column names default to "amplitude" and "sample_rate". Examples: >>> 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 numpy.ndarray(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. 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_providerrrrrrrrrrp)rrGr) rrrrrrrrrrrprrrVrVr] read_audios$E r) rrrrrinclude_timestampsrrrrrprrc  Cs0t|||t|||| |||d } t| | | | dS)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 numpy.ndarray(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. Returns: A :class:`~ray.data.Dataset` containing video frames from the video files. ) rrrrrrrrrrr)r)rGr)rrrrrrrrrrrrprrrVrVr] read_videoss&D r)pipelineschemarorrrpuridatabase collectionrrzpymongoarrow.api.Schemac Ks,td|||||d| } t| ||||dS)abCreate 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. 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. )rrrrrrNrV)rr) rrrrrrorrrp mongo_argsrrVrVr] read_mongols Sr project_iddatasetquerycCs t|||d}t|||||dS)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. 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)rrrrorrrprrVrVr] read_bigquerysArhive) rcolumnsrortensor_column_schemarrrrrrrrprr.rc Kst|t| |durt}t|fi|}|dd}|dd}|dd}t||||||||||| | | d }t|||| | dS)aACreates 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. 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. Ndataset_kwargs _block_udfr) rrto_batch_kwargsrrrrrrrrrr)'_emit_meta_provider_deprecation_warningrFrI_resolve_parquet_argspopr"r)rrrrorrrrrrrrrrparrow_parquet_argsrrrrrVrVr] read_parquetsF   rbeta)rrorrarrow_open_file_argsrrsizemoderrrrrrprrrcCsHt||dur t}t||| | |||||| | | d }t|||||dS)aCreates a :class:`~ray.data.Dataset` from image files. The column name defaults to "image". Examples: >>> import ray >>> path = "s3://anonymous@ray-example-data/batoidea/JPEGImages/" >>> ds = ray.data.read_images(path) >>> ds.schema() Column Type ------ ---- image numpy.ndarray(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 numpy.ndarray(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 numpy.ndarray(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. 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)rrrorrrrrrrrrrrrrprrVrVr] read_imagess2r) rrrorrrrrrrrrrpc  Ksnt|tdt|durt}t|fi|}|dur!||d<t||||||| | | d }t|||| | dS)aoCreate :class:`~ray.data.Dataset` from parquet files without reading metadata. Use :meth:`~ray.data.read_parquet` for most cases. Use :meth:`~ray.data.read_parquet_bulk` if all the provided paths point to files and metadata fetching using :meth:`~ray.data.read_parquet` takes too long or the parquet files do not all have a unified schema. Performance slowdowns are possible when using this method with parquet files that are very large. .. warning:: Only provide file paths as input (i.e., no directory paths). An OSError is raised if one or more paths point to directories. If your use-case requires directory paths, use :meth:`~ray.data.read_parquet` instead. Examples: Read multiple local files. You should always provide only input file paths (i.e. no directory paths) when known to minimize read latency. >>> ray.data.read_parquet_bulk( # doctest: +SKIP ... ["/path/to/file1", "/path/to/file2"]) Args: paths: A single file path or a list of file paths. 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. 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. 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. 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: [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 "*.parquet*". 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 paths. za`read_parquet_bulk` is deprecated and will be removed after May 2025. Use `read_parquet` instead.Nr)read_table_argsrrrrrrrr)rwarningswarnDeprecationWarningrHrr!r)rrrrorrrrrrrrrrprrrrVrVr]read_parquet_bulkks@e r)linesrrorrrrrrrrrrrprc Kst||r|||d}|D]\}}|rtd|dq|dur&t}t|||||| | | | d }|rHtjjj j }t |fd|i|}n t |fd|i|}t |||| |dS) aCreates 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. 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)rrqrxrGdictrXrcontextr=rdrrrr)rrrrorrrrrrrrrrrprincompatible_paramsparamvaluefile_based_datasource_kwargsrrrVrVr] read_jsons^z r) rrorrrrrrrrrrrpc  KsFt||dur t}t|||||||| | || d }t|||| | dS)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. 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)rrGrr)rrrorrrrrrrrrrrprrrVrVr]read_csvs0 rzutf-8T)encodingdrop_empty_linesrrorrrrrrrrrrrprrcCsHt||dur t}t|||||||| | | | | d }t|||||dS)a Create 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. 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)rrGr%r)rrrrrorrrrrrrrrrrprrVrVr] read_textKs0Yrc  CsDt||dur t}t||||||| | || d }t|||| | dS)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. 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)rrGrr)rrrorrrrrrrrrrrprrVrVr] read_avros,S r) rrorrrrrrrrrrpc  KsDt||dur t}t|| ||||||| || d }t||| | dS)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)rrGr r)rrrorrrrrrrrrrprrrVrVr] read_numpy-s,K r)rrorrrrrr tf_schemarrrrptfx_read_optionsrzschema_pb2.SchemarrRc Csddl}t|d}|r+|dkr+zddl}d}Wnty*d}tdYnw|dur2t}t|| |||||| || |d }t |||| | d}|r]|j r]|r]| s]dd 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. 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.) rrrrrrrrrrr)_infer_schema_and_transform) platformr processortfx_bslModuleNotFoundErrorloggerwarningrGr&rauto_infer_schema2ray.data._internal.datasource.tfrecords_datasourcer)rrrorrrrrrrrrrrprrtfx_readrrdsrrVrVr]read_tfrecordss\j     r )rrorrrdecoder fileselect filerenamesuffixes verbose_openrrrrrp expand_jsonr r r rrrcCsJt||dur t}t||||| | ||||| | | |d}t||||dS)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) r r r rrrrrrrrrrr)rrGr*r)rrrorrrr r r rrrrrrrprrrVrVr]read_webdataset6s2Fr) rrrorrrrrrrrrrpc  CsDt||dur t}t|||||||| | | d }t|||| | dS)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. ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. 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)rrGrr)rrrrorrrrrrrrrrprrVrVr]read_binary_filess,[ rMD5) shard_keys shard_hash_fnrorrrpsqlconnection_factoryrrc CsPt||||d}|r|dkr|durtd||stdt|||||dS)a Read 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. 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. )rrrrrsNz8shard_keys must be provided when override_num_blocks > 1zHDatabase does not support sharding. Please set override_num_blocks to 1.r)r$rxsupports_shardingr) rrrrrorrrprrVrVr]read_sql s(b  r)tablercatalogrrorrrp warehouse_idrrc Cs,ddlm} dd} tjd} | stdtjd} | s@ddlm} | r<| jj  }| d } ntd |sTdd lm }|d dd}|shdd lm }|d dd}|durt|durttd|r{d|}|durtd| | | ||||d}t|||||dS)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. 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)DatabricksUCDatasourcecSsXtd}zddl}|}|dur||jddWSty#|ty+|w)NzNo dbutils module found.r user_globaldbutils) RuntimeErrorIPython get_ipythonns_table ImportErrorKeyError)no_dbutils_errorr!ip_shellrVrVr] get_dbutils s  z+read_databricks_tables..get_dbutilsDATABRICKS_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.)hosttokenrrrr)rrorrrp)6ray.data._internal.datasource.databricks_uc_datasourcerosenvirongetrxray.util.spark.utilsr+notebook entry_point getDbutils getContexttagsr-rcollectr)rrrrrrorrrprr(r/r.r+rr-rrVrVr]read_databricks_tables sX H      r;)storage_optionsrrrp table_urir<cCst||d}t||||dS)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", ... ) Args: table_uri: The URI of the Hudi table to read from. Local file paths, S3, and GCS are supported. 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 `_. 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. )r=r<rrrrp)rr)r=r<rrrprrVrVr] read_hudi s'r?dfzdaft.DataFramecCs.t}|dus 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_versionr to_ray_dataset)r@pyarrow_versionrVrVr] from_daftR s   rDzdask.dataframe.DataFramecsZddl}ddlm}|}|j|d|i}ddlfddtfdd|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 schedulercs8t|jr t|St|tjr|Stdt|)Nz5Expected a Ray object ref or a Pandas DataFrame, got )r DataFramerXrYrKrxtype)r@)pandasrVr]to_ref s    zfrom_dask..to_refcs"g|] }tt|jqSrV)nextiterdaskvalues)r[part)rJrVr]r^ s"zfrom_dask..)rM ray.util.daskrE to_delayedpersistrIfrom_pandas_refs)r@rMrE partitionspersisted_partitionsr rV)rIrJr] from_daskq s   rVzmars.dataframe.DataFramecCsddlm}||}|S)aMCreate a :class:`~ray.data.Dataset` from a `Mars DataFrame `_. 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 dataframerB)r@mdr rVrVr] from_mars s rZz modin.pandas.dataframe.DataFramecCs$ddlm}||dd}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.partitionsr[rS)r@r[partsr rVrVr] from_modin s r`dfszpandas.DataFramecsddl}t||jr |g}|dur)t|dkr|j|dd}n|d}t||}ddlmt }|j r?fdd|D}t dd|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. rNrsr\)_cast_ndarray_columns_to_tensor_extensioncsg|]}|qSrV)rer[r@rbrVr]r^ zfrom_pandas..cSrUrVrWrdrVrVr]r^ r_) rIrrGr|concatnp array_split"ray.air.util.data_batch_conversionrcr=rdenable_tensor_extension_castingrS)rarppdaryrrVrbr] from_pandas s    rmcs6t|tjr |g}n#t|tr$|D]}t|tjs"tdt|qn tdt|t}|jrat t t fdd|D}t t d|iddt}tt|||j}t||St tdd fd d|D}ttt|\}}t |}t t d|iddt}tt|||j}t||S) a6Create a :class:`~ray.data.Dataset` from a list of Ray object references to pandas dataframes. Examples: >>> 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 cg|]}|qSrVremoterdget_metadata_schemarVr]r^ r_z$from_pandas_refs..r1Nra num_returnscrprVrqrd) df_to_blockrVr]r^( r_)rrXrKlistrxrHr=rdenable_pandas_blockr4r7r3r3r5rer,r1rfr?r9mapzip)rar@rmetadata_schemarjrkresrTrV)rxrtr]rS sZ          rSndarrayscCs$t|tjr |g}tdd|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. cSrUrVrWr[ndarrayrVrVr]r^S r_zfrom_numpy..)rrgrfrom_numpy_refs)rrVrVr] from_numpy8 s rcst|tjr |g}n#t|tr$|D]}t|tjs"tdt|qn tdt|ttt ddfdd|D}t tt |\}}t |}t td|idd t}tt|||j}t||S) a%Creates a :class:`~ray.data.Dataset` from a list of Ray object references to NumPy ndarrays. The column name defaults to "data". Examples: >>> 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. rnrorurvcsg|]}|qSrVrqrrndarray_to_block_remoterVr]r^ rez#from_numpy_refs..r0Nra)rrXrKryrxrHr=rdr4r8r{r|r3r3r5rer,r0rfr?)rrr~rTr}rjrkrVrr]rV s>        rrptables pyarrow.Tablec s$ddl}ddl}t||jtfr|g}|dur|dkrtdt|dkr*||n|dt}|dkrCfdd||D}nF||d|}g}||D]}||}||kr^nt |||} | || qRt||kr dd} | | g|t||}t dd|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 > 0rscsg|]}ddqS)r)slice)r[rcombined_tablerVr]r^ s zfrom_arrow..cSrUrVrWr[trVrVr]r^ r_)rvpyarrowrTablebytesrxr| concat_tablesr~r{rrextendfrom_arrow_refs) rrprvpa total_rows batch_sizeslicesrstartlength empty_tablerVrr] from_arrow s4    rcslt|tjr |g}tttfdd|D}ttd|iddt }t t |||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. crprVrqrrsrVr]r^ r_z#from_arrow_refs..r-Nra)rrXrKr4r7r3r3r5r=rdrer,r-rfr?)rr}rjrkrVrsr]r s   r)limitversion timestampjson_predicate_hintsrrrpurlrrrrc Cs&t|||||d}tjj||||dS)a` Read data from a Delta Sharing table. Delta Sharing projct https://github.com/delta-io/delta-sharing/tree/main This function reads data from a Delta Sharing table specified by the URL. It supports various options such as limiting the number of rows, specifying a version or timestamp, and configuring concurrency. Before calling this function, ensure that the URL is correctly formatted to point to the Delta Sharing table you want to access. Make sure you have a valid delta_share profile in the working directory. Examples: .. testcode:: :skipif: True import ray ds = ray.data.read_delta_sharing_tables( url=f"your-profile.json#your-share-name.your-schema-name.your-table-name", limit=100000, version=1, ) Args: url: A URL under the format "#..". 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. 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. )rrrrrr>)rrXrr) rrrrrrrrprrVrVr]read_delta_sharing_tables sE rzpyspark.sql.DataFramecCs 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)raydprwsparkspark_dataframe_to_ray_dataset)r@rorprrVrVr] from_sparkd s r)zdatasets.Datasetzdatasets.IterableDatasetc Csddl}ddlm}ddlm}t||j|jfrzy||}t |dkrddl }g} |D]D} z#|j | ddd} | j dkrD| | jn td | j d | d Wq-|jyq} ztd | d | d WYd} ~ q-d} ~ ww| sxtdddl} | jj}t| ||||dt|gidWSWnt|fytdYnwt||jrt||d|||dSt||jr|d}t|dd|d}|St||j|jfrt|}td|dd|dt dt!|)a Create a :class:`~ray.data.MaterializedDataset` from a `Hugging Face Datasets Dataset `_ or a :class:`~ray.data.Dataset` from a `Hugging Face Datasets IterableDataset `_. For an `IterableDataset`, we use a streaming implementation to read data. If the dataset is a public Hugging Face Dataset that is hosted on the Hugging Face Hub and no transformations have been applied, then the `hosted parquet files `_ will be passed to :meth:`~ray.data.read_parquet` to perform a distributed read. All other cases will be done with a single node read. Example: .. The following `testoutput` is mocked to avoid illustrating download logs like "Downloading and preparing dataset 162.17 MiB". .. testcode:: import ray import datasets hf_dataset = datasets.load_dataset("tweet_eval", "emotion") ray_ds = ray.data.from_huggingface(hf_dataset["train"]) print(ray_ds) hf_dataset_stream = datasets.load_dataset("tweet_eval", "emotion", streaming=True) ray_ds_stream = ray.data.from_huggingface(hf_dataset_stream["train"]) print(ray_ds_stream) .. testoutput:: :options: +MOCK MaterializedDataset( num_blocks=..., num_rows=3257, schema={text: string, label: int64} ) Dataset( num_rows=3257, schema={text: string, label: int64} ) 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)rorrrprz]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_datasourcerrIterableDatasetr>list_parquet_urls_from_datasetr|requestshead status_coderrrrRequestExceptionFileNotFoundErrorfsspec.implementations.httpimplementationshttpHTTPFileSystemrr with_formatr DatasetDictIterableDatasetDictrykeysr TypeErrorrH)rrorrprrr file_urlsr resolved_urlsrrespefsspecr hf_ds_arrowray_dsavailable_keysrVrVr]from_huggingface sC      +     rztf.data.DatasetcCstt|S)aCreate a :class:`~ray.data.Dataset` from a `TensorFlow Dataset `_. 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: numpy.ndarray(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`_. )rryas_numpy_iteratorrrVrVr]from_tf! s9rztorch.utils.data.Dataset local_readcCs6i}|rttdddd}tt|d|ddS)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)rnum_cpusrrs)rrp)rOrXrrrr')rrrrVrVr] from_torch] s!  r)*) row_filterroselected_fields snapshot_id scan_kwargscatalog_kwargsrrptable_identifierrrPrrrrc Cs(t||||||d} t| |||d} | 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 pyiceberg.expressions import EqualTo #doctest: +SKIP >>> ds = ray.data.read_iceberg( #doctest: +SKIP ... table_identifier="db_name.table_name", ... row_filter=EqualTo("column_name", "literal_value"), ... catalog_kwargs={"name": "default", "type": "glue"} ... ) Args: table_identifier: Fully qualified table identifier (``db_name.table_name``) row_filter: 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: 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. 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. )rrrrrr)rrorpr)rr) rrrorrrrrrprrrVrVr] read_iceberg sB r)rfilterr<scanner_optionsrrrprrc Cs"t|||||d}t||||dS)aZ 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. 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. 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. )rrrr<rr>)rr) rrrr<rrrrprrVrVr] read_lance s3r)rrorder_byclient_settings client_kwargsrrrpdsnrrrc Cs&t|||||||d} t| ||| dS)a 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. 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. )rrrrrrrr>)rr) rrrrrrrrrrprrVrVr]read_clickhouse+s@ r) data_formatregionr/rr reader_kwargscCst||||||d}|S)a Loads 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](https://docs.databricks.com/en/data-governance/unity-catalog/credential-vending.html)), 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 ``read_unity_catalog`` function is currently experimental and under active development .. warning:: The Databricks Unity Catalog credential vending feature is currently in Public Preview and there are important requirements and limitations. You must read these docs carefully and ensure your workspace and principal are properly configured. Features: - **Secure Access**: Only principals with `EXTERNAL USE SCHEMA` on the containing schema, and after explicit metastore enablement, can obtain short-lived credentials. - **Format Support**: Supports reading `delta` and `parquet` formats via supported Ray Dataset readers (iceberg coming soon). - **Cloud Support**: AWS, Azure, and GCP supported, with automatic environment setup for the vended credentials per session. - **Auto-Infer**: Data format is auto-inferred from table metadata, but can be explicitly specified. Examples: Read a Unity Catalog managed Delta table with credential vending: >>> import ray >>> ds = read_unity_catalog( # doctest: +SKIP ... table="main.sales.transactions", ... url="https://dbc-XXXXXXX-XXXX.cloud.databricks.com", # noqa: E501 ... token="XXXXXXXXXXX" # noqa: E501 ... ) >>> ds.show(3) # doctest: +SKIP Explicitly specify the format, and pass reader options: >>> ds = read_unity_catalog( # doctest: +SKIP ... table="main.catalog.images", ... url="https://dbc-XXXXXXX-XXXX.cloud.databricks.com", # noqa: E501 ... token="XXXXXXXXXXX", # noqa: E501 ... data_format="delta", ... region="us-west-2", ... # Reader kwargs come from the associated reader (ray.data.read_delta in this example) ... reader_kwargs={"override_num_blocks": 1000} ... ) Args: table: Unity Catalog table name as `..`. Must be a managed or external table supporting credential vending. url: Databricks workspace URL, e.g. `"https://dbc-XXXXXXX-XXXX.cloud.databricks.com"` token: Databricks PAT (Personal Access Token) with `EXTERNAL USE SCHEMA` on the schema containing the table, and with access to the workspace API. data_format: (Optional) Data format override. If not specified, inferred from Unity Catalog metadata and file extension. Supported: `"delta"`, `"parquet"` region: (Optional) For S3: AWS region for cloud credential environment setup. reader_kwargs: Additional arguments forwarded to the underlying Ray Dataset reader (e.g., override_num_blocks, etc.). Returns: A :class:`ray.data.Dataset` containing the data from the external Unity Catalog table. References: - Databricks Credential Vending: https://docs.databricks.com/en/data-governance/unity-catalog/credential-vending.html - API Reference for temporary credentials: https://docs.databricks.com/api/workspace/unity-catalog/temporary-table-credentials )base_urlr/table_full_namerrr)r(read)rrr/rrrreaderrVrVr]read_unity_catalog}sJr) rrrorrrrrrrrppathc Ksddl} d}z| |Wnty%td|d|d|d|d wdd lm}t|ts5td ||}d g}t |f||||||||| || | d | S) a6 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. 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. 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.parquet) rrrorrrrrrrrrp) importlib import_moduler$rrrstrrx file_urisr)rrrrorrrrrrrrprrpackagerrrrVrVr] read_deltasHG    rr rkwargscCs8t||jrtdt|jdi|}|S|}|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`.NrV)r= _set_currentshould_create_readerrrr create_reader)r rrrrVrVr]rBs rc s0dur|dddfdd }||d<|S)Nrr\rrrc spddlm}D]#\}\tfdd||D}|||||||}q dur6|}|S)Nr)ArrowTensorArraycs g|] }tj|dqS))bufferdtype)rgr as_buffer)r[bufrrrVr]r^qsz=_resolve_parquet_args.._block_udf..)ray.data.extensionsrrqrcolumn set_column_ensure_integer_indexr)r\rtensor_col_namenp_colexisting_block_udfrrr]ris   z)_resolve_parquet_args.._block_udf)r\rrrr)r)rrrrVrr]rbs  rcCs&|dkr td|S|dur|}|S)NrmzpThe argument ``parallelism`` is deprecated in Ray 2.10. Please specify argument ``override_num_blocks`` instead.)rrrnrVrVr]rwsrwcCs|dur tdtdSdS)NzNThe `meta_provider` argument is deprecated and will be removed after May 2025.)rrr)rrVrVr]rs r)NN)N)rmNN)F)rmN)rloggingr1rtypingrrrrrrrr r r numpyrgpackaging.versionr rArXray._private.arrow_utilsr ray._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_datasourcer0ray.data._internal.datasource.iceberg_datasourcer.ray.data._internal.datasource.image_datasourcerr-ray.data._internal.datasource.json_datasourcerrr.ray.data._internal.datasource.lance_datasourcer.ray.data._internal.datasource.mongo_datasourcer.ray.data._internal.datasource.numpy_datasourcer 5ray.data._internal.datasource.parquet_bulk_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%rr&.ray.data._internal.datasource.torch_datasourcer'6ray.data._internal.datasource.unity_catalog_datasourcer(.ray.data._internal.datasource.video_datasourcer)3ray.data._internal.datasource.webdataset_datasourcer*+ray.data._internal.delegating_block_builderr+%ray.data._internal.logical.interfacesr,3ray.data._internal.logical.operators.from_operatorsr-r.r/r0r12ray.data._internal.logical.operators.read_operatorr2ray.data._internal.planr3ray.data._internal.remote_fnr4ray.data._internal.statsr5ray.data._internal.utilr6r7r8r9ray.data.blockr:r;r<ray.data.contextr=ray.data.datasetr>r?ray.data.datasourcer@rArBrCray.data.datasource.datasourcerD)ray.data.datasource.file_based_datasourcerErF&ray.data.datasource.file_meta_providerrGrH)ray.data.datasource.parquet_meta_providerrI ray.data.datasource.partitioningrJ ray.typesrKray.util.annotationsrLrMrNray.util.scheduling_strategiesrOdaftrMrmarsmodinrIrpymongoarrow.api pymongoarrowpyspark tensorflowtftorchpyiceberg.expressionsrPtensorflow_metadata.proto.v0rQrRrS getLogger__name__rrlintrr~rrr_FILE_EXTENSIONSboolrrrrrrrrrrrrrr callableryrrrr;r?rDrVrZr`rmrSrrrrrrrrrrrrrrrrrrrrwrrVrVrVr]s 0                                           T 3 : X     X      X     c  J         3                          -       )       t      l      d       $       b       t    y    3# 5 I$ ? F 0  U   " ; 5     T   B     QT       o !  $