python/ray/data/datasource/file_based_datasource.py
def prepare_read(
        self,
        parallelism: int,
        paths: Union[str, List[str]],
        filesystem: Optional["pyarrow.fs.FileSystem"] = None,
        schema: Optional[Union[type, "pyarrow.lib.Schema"]] = None,
        open_stream_args: Optional[Dict[str, Any]] = None,
        meta_provider: BaseFileMetadataProvider = DefaultFileMetadataProvider(),
        partition_filter: PathPartitionFilter = None,
        # TODO(ekl) deprecate this once read fusion is available.
        _block_udf: Optional[Callable[[Block], Block]] = None,
        **reader_args,
    ) -> List[ReadTask]:
        """Creates and returns read tasks for a file-based datasource."""
        _check_pyarrow_version()
        import numpy as np

        paths, filesystem = _resolve_paths_and_filesystem(paths, filesystem)
        paths, file_sizes = meta_provider.expand_paths(paths, filesystem)
        if partition_filter is not None:
            paths = partition_filter(paths)

        read_stream = self._read_stream

        filesystem = _wrap_s3_serialization_workaround(filesystem)

        if open_stream_args is None:
            open_stream_args = {}

        def read_files(
            read_paths: List[str],
            fs: Union["pyarrow.fs.FileSystem", _S3FileSystemWrapper],
        ) -> Iterable[Block]:
            logger.debug(f"Reading {len(read_paths)} files.")
            if isinstance(fs, _S3FileSystemWrapper):
                fs = fs.unwrap()
            ctx = DatasetContext.get_current()
            output_buffer = BlockOutputBuffer(
                block_udf=_block_udf, target_max_block_size=ctx.target_max_block_size
            )
            for read_path in read_paths:
                compression = open_stream_args.pop("compression", None)
                if compression is None:
                    import pyarrow as pa

                    try:
                        # If no compression manually given, try to detect
                        # compression codec from path.
                        compression = pa.Codec.detect(read_path).name
                    except (ValueError, TypeError):
                        # Arrow's compression inference on the file path
                        # doesn't work for Snappy, so we double-check ourselves.
                        import pathlib

                        suffix = pathlib.Path(read_path).suffix
                        if suffix and suffix[1:] == "snappy":
                            compression = "snappy"
                        else:
                            compression = None
                if compression == "snappy":
                    # Pass Snappy compression as a reader arg, so datasource subclasses
                    # can manually handle streaming decompression in
                    # self._read_stream().
                    reader_args["compression"] = compression
                    reader_args["filesystem"] = fs
                elif compression is not None:
                    # Non-Snappy compression, pass as open_input_stream() arg so Arrow
                    # can take care of streaming decompression for us.
                    open_stream_args["compression"] = compression
                with fs.open_input_stream(read_path, **open_stream_args) as f:
                    for data in read_stream(f, read_path, **reader_args):
                        output_buffer.add_block(data)
                        if output_buffer.has_next():
                            yield output_buffer.next()
            output_buffer.finalize()
            if output_buffer.has_next():
                yield output_buffer.next()

        read_tasks = []
        for read_paths, file_sizes in zip(
            np.array_split(paths, parallelism), np.array_split(file_sizes, parallelism)
        ):
            if len(read_paths) <= 0:
                continue

            meta = meta_provider(
                read_paths,
                schema,
                rows_per_file=self._rows_per_file(),
                file_sizes=file_sizes,
            )
            read_task = ReadTask(
                lambda read_paths=read_paths: read_files(read_paths, filesystem), meta
            )
            read_tasks.append(read_task)

        return read_tasks
Creates and returns read tasks for a file-based datasource.
Creates and returns read tasks for a file-based datasource.
Creates and returns read tasks for a file - based datasource .
def prepare_read(
        self,
        parallelism: int,
        paths: Union[str, List[str]],
        filesystem: Optional["pyarrow.fs.FileSystem"] = None,
        schema: Optional[Union[type, "pyarrow.lib.Schema"]] = None,
        open_stream_args: Optional[Dict[str, Any]] = None,
        meta_provider: BaseFileMetadataProvider = DefaultFileMetadataProvider(),
        partition_filter: PathPartitionFilter = None,
        _block_udf: Optional[Callable[[Block], Block]] = None,
        **reader_args,
    ) -> List[ReadTask]:
        _check_pyarrow_version()
        import numpy as np
        paths, filesystem = _resolve_paths_and_filesystem(paths, filesystem)
        paths, file_sizes = meta_provider.expand_paths(paths, filesystem)
        if partition_filter is not None:
            paths = partition_filter(paths)
        read_stream = self._read_stream
        filesystem = _wrap_s3_serialization_workaround(filesystem)
        if open_stream_args is None:
            open_stream_args = {}
        def read_files(
            read_paths: List[str],
            fs: Union["pyarrow.fs.FileSystem", _S3FileSystemWrapper],
        ) -> Iterable[Block]:
            logger.debug(f"Reading {len(read_paths)} files.")
            if isinstance(fs, _S3FileSystemWrapper):
                fs = fs.unwrap()
            ctx = DatasetContext.get_current()
            output_buffer = BlockOutputBuffer(
                block_udf=_block_udf, target_max_block_size=ctx.target_max_block_size
            )
            for read_path in read_paths:
                compression = open_stream_args.pop("compression", None)
                if compression is None:
                    import pyarrow as pa
                    try:
                        compression = pa.Codec.detect(read_path).name
                    except (ValueError, TypeError):
                        import pathlib
                        suffix = pathlib.Path(read_path).suffix
                        if suffix and suffix[1:] == "snappy":
                            compression = "snappy"
                        else:
                            compression = None
                if compression == "snappy":
                    reader_args["compression"] = compression
                    reader_args["filesystem"] = fs
                elif compression is not None:
                    open_stream_args["compression"] = compression
                with fs.open_input_stream(read_path, **open_stream_args) as f:
                    for data in read_stream(f, read_path, **reader_args):
                        output_buffer.add_block(data)
                        if output_buffer.has_next():
                            yield output_buffer.next()
            output_buffer.finalize()
            if output_buffer.has_next():
                yield output_buffer.next()
        read_tasks = []
        for read_paths, file_sizes in zip(
            np.array_split(paths, parallelism), np.array_split(file_sizes, parallelism)
        ):
            if len(read_paths) <= 0:
                continue
            meta = meta_provider(
                read_paths,
                schema,
                rows_per_file=self._rows_per_file(),
                file_sizes=file_sizes,
            )
            read_task = ReadTask(
                lambda read_paths=read_paths: read_files(read_paths, filesystem), meta
            )
            read_tasks.append(read_task)
        return read_tasks
def prepare_read ( self , parallelism : int , paths : Union [ str , List [ str ] ] , filesystem : Optional [ "pyarrow.fs.FileSystem" ] = None , schema : Optional [ Union [ type , "pyarrow.lib.Schema" ] ] = None , open_stream_args : Optional [ Dict [ str , Any ] ] = None , meta_provider : BaseFileMetadataProvider = DefaultFileMetadataProvider ( ) , partition_filter : PathPartitionFilter = None , _block_udf : Optional [ Callable [ [ Block ] , Block ] ] = None , ** reader_args , ) -> List [ ReadTask ] : _check_pyarrow_version ( ) import numpy as np paths , filesystem = _resolve_paths_and_filesystem ( paths , filesystem ) paths , file_sizes = meta_provider . expand_paths ( paths , filesystem ) if partition_filter is not None : paths = partition_filter ( paths ) read_stream = self . _read_stream filesystem = _wrap_s3_serialization_workaround ( filesystem ) if open_stream_args is None : open_stream_args = { } def read_files ( read_paths : List [ str ] , fs : Union [ "pyarrow.fs.FileSystem" , _S3FileSystemWrapper ] , ) -> Iterable [ Block ] : logger . debug ( f"Reading {len(read_paths)} files." ) if isinstance ( fs , _S3FileSystemWrapper ) : fs = fs . unwrap ( ) ctx = DatasetContext . get_current ( ) output_buffer = BlockOutputBuffer ( block_udf = _block_udf , target_max_block_size = ctx . target_max_block_size ) for read_path in read_paths : compression = open_stream_args . pop ( "compression" , None ) if compression is None : import pyarrow as pa try : compression = pa . Codec . detect ( read_path ) . name except ( ValueError , TypeError ) : import pathlib suffix = pathlib . Path ( read_path ) . suffix if suffix and suffix [ 1 : ] == "snappy" : compression = "snappy" else : compression = None if compression == "snappy" : reader_args [ "compression" ] = compression reader_args [ "filesystem" ] = fs elif compression is not None : open_stream_args [ "compression" ] = compression with fs . open_input_stream ( read_path , ** open_stream_args ) as f : for data in read_stream ( f , read_path , ** reader_args ) : output_buffer . add_block ( data ) if output_buffer . has_next ( ) : yield output_buffer . next ( ) output_buffer . finalize ( ) if output_buffer . has_next ( ) : yield output_buffer . next ( ) read_tasks = [ ] for read_paths , file_sizes in zip ( np . array_split ( paths , parallelism ) , np . array_split ( file_sizes , parallelism ) ) : if len ( read_paths ) <= 0 : continue meta = meta_provider ( read_paths , schema , rows_per_file = self . _rows_per_file ( ) , file_sizes = file_sizes , ) read_task = ReadTask ( lambda read_paths = read_paths : read_files ( read_paths , filesystem ) , meta ) read_tasks . append ( read_task ) return read_tasks
Creates and returns read tasks for a file-based datasource.
Creates and returns read tasks for a file - based datasource .
# TODO(ekl) deprecate this once read fusion is available. """Creates and returns read tasks for a file-based datasource.""" # If no compression manually given, try to detect # compression codec from path. # Arrow's compression inference on the file path # doesn't work for Snappy, so we double-check ourselves. # Pass Snappy compression as a reader arg, so datasource subclasses # can manually handle streaming decompression in # self._read_stream(). # Non-Snappy compression, pass as open_input_stream() arg so Arrow # can take care of streaming decompression for us.
self
parallelism
int
paths
Union[str, List[str]]
filesystem
Optional["pyarrow.fs.FileSystem"]
schema
Optional[Union[type, "pyarrow.lib.Schema"]]
open_stream_args
Optional[Dict[str, Any]]
meta_provider
BaseFileMetadataProvider
partition_filter
PathPartitionFilter
_block_udf
Optional[Callable[[Block], Block]]