o $i|M@sddlZddlZddlmZddlmZmZmZmZm Z m Z m Z m Z m Z mZddlZddlZddlmZmZmZmZmZmZmZddlmZmZddlmZddlm Z m!Z!ddl"m#Z#m$Z$dd l%m&Z&m'Z'm(Z(dd l)m*Z*m+Z+dd l,m-Z-erddl.Z/ddl0Z0e1e2Z3d Z4d Z5e-eGd ddZ6e-Gddde Z7deddee8efdedfddZ9dddee8efddfddZ:dddee8efddfddZ;d/d!d"Zd(egee8effdee8effd)d*Z?d+ee d,e6dfddfd-d.Z@dS)0N) dataclass) TYPE_CHECKINGAnyCallableDictIterableIteratorListLiteralOptionalUnion)RetryingContextManagerRetryingPyFileSystem_check_pyarrow_version_is_local_schemeinfer_compressioniterate_with_retrymake_async_gen)Block BlockAccessor) DataContext) DatasourceReadTask)BaseFileMetadataProviderDefaultFileMetadataProvider) PartitioningPathPartitionFilterPathPartitionParser)_has_file_extension_resolve_paths_and_filesystem) DeveloperAPIc@s*eZdZUdZdZeeed<ddZdS)FileShuffleConfiga^Configuration for file shuffling. This configuration object controls how files are shuffled while reading file-based datasets. .. note:: Even if you provided a seed, you might still observe a non-deterministic row order. This is because tasks are executed in parallel and their completion order might vary. If you need to preserve the order of rows, set `DataContext.get_current().execution_options.preserve_order`. Args: seed: An optional integer seed for the file shuffler. If provided, Ray Data shuffles files deterministically based on this seed. Example: >>> import ray >>> from ray.data import FileShuffleConfig >>> shuffle = FileShuffleConfig(seed=42) >>> ds = ray.data.read_images("s3://anonymous@ray-example-data/batoidea", shuffle=shuffle) NseedcCs&|jdurt|jtstddSdS)z2Ensure that the seed is either None or an integer.Nz Seed must be an integer or None.)r# isinstanceint ValueErrorselfr)f/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/datasource/file_based_datasource.py __post_init__YszFileShuffleConfig.__post_init__) __name__ __module__ __qualname____doc__r#r r%__annotations__r+r)r)r)r*r">s  r"cseZdZUdZdZdZeeee efe d<dZ ddde ddddddd deee efded d eee d fd eeeefd ededededeeedefdedee effddZde efddZde efddZdeefddZ d7dedeede efd d!Zd"ed#eeefdeefd$d%Zd#eeefdeefd&d'Z ( )  *  +d8d,d-Z dd*d"edd)fd.d/Z!d0d1Z"d2d)d"ede#e$fd3d4Z%e&defd5d6Z'Z(S)9FileBasedDatasourcezFile-based datasource for reading files. Don't use this class directly. Instead, subclass it and implement `_read_stream()`. FN_FILE_EXTENSIONSr) filesystemschemaopen_stream_args meta_providerpartition_filter partitioningignore_missing_pathsshuffle include_pathsfile_extensionspathsr3pyarrow.fs.FileSystemr4zpyarrow.lib.Schemar5r6r7r8r9r:filesr;r<c sttt| |_|jstjjjrt d||_ t |_ ||_||_||_||_||_| |_||_t||\}|_tj|j|j jd|_ttt|j||j||d\}} |rit|dkrit d|jdurtt|| ||}fdd|D} t|dkrt ddurtt|| fd d|D}fd d|D} t|dkrt d d t | d|_!| d krt"j#$|_!n t%| t&rt"j#$| j'|_!t(||_)t(| |_*dS)NzBecause you're using Ray Client, read tasks scheduled on the Ray cluster can't access your local files. To fix this issue, store files in cloud storage or a distributed filesystem like NFS.)retryable_errors)r9rzRNone of the provided paths exist. The 'ignore_missing_paths' field is set to True.cg|]}|qSr)r).0p path_to_sizer)r* z0FileBasedDatasource.__init__..z`No input files found to read. Please double check that 'partition_filter' field is set properly.csg|] }t|r|qSr))rrB)r<r)r*rGscrAr)r)rBrEr)r*rGrHzANo input files found to read with the following file extensions: zC. Please double check that 'file_extensions' field is set properly.r?)+super__init__rr_supports_distributed_readsrayutilclient is_connectedr&_schemar get_current _data_context_open_stream_args_meta_provider_partition_filter _partitioning_ignore_missing_paths_include_paths _source_pathsr _filesystemrwrapretried_io_errorsmaplistzip expand_pathslendict_validate_shuffle_arg_file_metadata_shufflernprandom default_rngr$r"r#put _paths_ref_file_sizes_ref) r(r=r3r4r5r6r7r8r9r:r;r< file_sizes __class__)r<rFr*rJosz           zFileBasedDatasource.__init__returncC t|jSN)rLgetrir'r)r)r*_paths zFileBasedDatasource._pathscCrorp)rLrqrjr'r)r)r* _file_sizesrszFileBasedDatasource._file_sizescCs&d}|D] }|dur||7}q|SNr)rt)r( total_sizeszr)r)r*estimate_inmemory_data_sizes  z/FileBasedDatasource.estimate_inmemory_data_size parallelismper_task_row_limitcs@ddl}jj}}jdur8tt||fddjt D}tt tt|\}}t j durCidt tdt tffdd fdd }t|t |}g}|||} |||} t| | D]'\} }t | dkrqvj| |d } || j} t| | |d }||qv|S) NrcrAr)r))rCi)files_metadatar)r*rGsz6FileBasedDatasource.get_read_tasks.. read_pathsrnc 3st}|D]Yi}durt}|}tj|fijd2tfdddjjdD]}|r?t||}jrMt |}| d}|Vq6Wdn1s[wYqdS)N)contextcs Srp) _read_streamr))f read_pathr(r)r*s zHFileBasedDatasource.get_read_tasks..read_files..zread stream iteratively) descriptionmatchpath) #_unwrap_s3_serialization_workaroundrr _open_input_sourcerRrr\_add_partitionsrXr for_block fill_column)r}fs partitionsparseblockblock_accessor)r3r5r8r()rrr* read_filess6    z6FileBasedDatasource.get_read_tasks..read_filescsfdd}|S)Nc3sjjjrddkr0tttdtddttddEdHdStdtdEdHdS)NrzReading z files with z threads.T) num_workerspreserve_orderingz files.) rRexecution_optionspreserve_orderminraloggerdebugriterr)) num_threadsrr}r(r)r* read_task_fns  zUFileBasedDatasource.get_read_tasks..create_read_task_fn..read_task_fnr))r}rr)rr()rr}r*create_read_task_fnsz?FileBasedDatasource.get_read_tasks..create_read_task_fn) rows_per_filerk)rz)numpyrSrVrrrtrdr^r_ permutationrar]!_wrap_s3_serialization_workaroundrZrstrrr array_splitrT_rows_per_file_NUM_THREADS_PER_TASKrappend)r(ryrzrer=rkshuffled_files_metadatar read_tasks split_pathssplit_file_sizesr}metar read_taskr))r|r3r5r8rr(r*get_read_taskssL        z"FileBasedDatasource.get_read_tasksr open_argscCs |dd}|durt|}|S)aResolves the compression format for a stream. Args: path: The file path to resolve compression for. open_args: kwargs passed to `pyarrow.fs.FileSystem.open_input_stream `_ when opening input files to read. Returns: The compression format (e.g., "gzip", "snappy", "bz2") or None if no compression is detected or specified. compressionN)rqr)r(rrrr)r)r*resolve_compressionFs z'FileBasedDatasource.resolve_compressioncCs |dd}|dur|jj}|S)N buffer_size)poprRstreaming_read_buffer_size)r(rrr)r)r*_resolve_buffer_sizeZs z(FileBasedDatasource._resolve_buffer_sizefilepyarrow.NativeFilerpyarrow.PythonFilecCsjddl}ddl}ddlm}t}t||r"|jj ||dn|j ||d| d|j |ddS)Nr)HadoopFileSystem)srcdstr)mode) pyarrowsnappy pyarrow.fsrioBytesIOr$unwrap hadoop_snappystream_decompressseek PythonFile)r(rr3parrstreamr)r)r*_file_to_snappy_stream`s  z*FileBasedDatasource._file_to_snappy_streamcKsf|||}||}|dkr$d|d<|j|fd|i|}|||S||d<|j|fd|i|S)aOpens a source path for reading and returns the associated Arrow NativeFile. The default implementation opens the source path as a sequential input stream, using self._data_context.streaming_read_buffer_size as the buffer size if none is given by the caller. Implementations that do not support streaming reads (e.g. that require random access) should override this method. rNrr)rropen_input_streamr)r(r3rrrrrr)r)r*rrs   z&FileBasedDatasource._open_input_sourcecCsdS)z8Returns the number of rows per file, or None if unknown.Nr)r'r)r)r*rsz"FileBasedDatasource._rows_per_filercCstd)z`Streaming read a single file. This method should be implemented by subclasses. z@Subclasses of FileBasedDatasource must implement _read_stream().)NotImplementedError)r(rrr)r)r*rsz FileBasedDatasource._read_streamcC|jSrp)rKr'r)r)r*supports_distributed_readssz.FileBasedDatasource.supports_distributed_readsrp)rrr3rrnr))r,r-r.r/_WRITE_FILE_PER_ROWr2r r rr r0rrtyperrrrrboolr r"rJrrfloatrtr%rxrrrrrrrrrrpropertyr __classcell__r)r)rlr*r1_s      ^ l      r1data) pyarrow.Table pd.DataFramerrncCsVddl}ddl}t||j|jfsJt||jrt||St||jr)t||SdSru)pandasrr$Table DataFrame_add_partitions_to_table_add_partitions_to_dataframe)rrpdrr)r)r*rs    rtablerc Csddl}ddlm}t|j}|D]W\}}||gt|}||vrd|j |j }| |}| | |||} | } | sVtd|d|d||d|j|} || ||}q|||}q|S)NrPartition column , exists in table data, but partition value '$' is different from in-data values: .)rpyarrow.computecomputeset column_namesitemsarrayrar4fieldrcastallequalas_pyr&unique to_pylistget_field_index set_column append_column) rrrpcrrvaluecolumn column_typevalues_are_equalr{r)r)r*rs,    rdfrc Csddl}|D]E\}}|j|gt||d}||vrI|||j}||}|||||sItd|d|dt || d|||<q|S)Nr)rnamerrrr) rrSeriesraastypedtypenotnaequalsr&r^r)rrrrrrmaskr)r)r*rs   rr3r>cCs@ddl}ddl}|}t|tr|}t||jjrt|S|Sru)rrr$rrr S3FileSystem_S3FileSystemWrapper)r3rrbase_fsr)r)r*rs r)r>rcCst|tr |}|Srp)r$rr)r3r)r)r*rs rc@s6eZdZdZd ddZddZedd Zd d Zd S)raxpyarrow.fs.S3FileSystem wrapper that can be deserialized safely. Importing pyarrow.fs during reconstruction triggers the pyarrow S3 subsystem initialization. NOTE: This is only needed for pyarrow<14.0.0 and should be removed once the minimum supported pyarrow version exceeds that. See https://github.com/apache/arrow/pull/38375 for context. rr>cCs ||_dSrp_fs)r(rr)r)r*rJ s z_S3FileSystemWrapper.__init__cCrrprr'r)r)r*rsz_S3FileSystemWrapper.unwrapcCsddl}|||Sru)r)clsfs_reconstructfs_argsrr)r)r* _reconstructs z!_S3FileSystemWrapper._reconstructcCstj|jfSrp)rrr __reduce__r'r)r)r*rsz_S3FileSystemWrapper.__reduce__N)rr>) r,r-r.r/rJr classmethodrrr)r)r)r*rs   r kwargs_fncKs|r |}|||Srp)update)rkwargskwarg_overridesr)r)r*_resolve_kwargss r r:r?cCs6|dus|dkst|tstd|ddSdSdS)Nr?zInvalid value for 'shuffle': z6. Valid values are None, 'files', `FileShuffleConfig`.)r$r"r&)r:r)r)r*rc&s   rc)r3r>)Arlogging dataclassesrtypingrrrrrrr r r r rrerLray.data._internal.utilr rrrrrrray.data.blockrrray.data.contextrray.data.datasource.datasourcerr&ray.data.datasource.file_meta_providerrr ray.data.datasource.partitioningrrrray.data.datasource.path_utilrrray.util.annotationsr rrr getLoggerr,r)FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLDPATHS_PER_FILE_SIZE_FETCH_TASKr"r1rrrrrrrr rcr)r)r)r*s~ 0 $   F