python/ray/data/datasource/file_based_datasource.py
def do_write(
        self,
        blocks: List[ObjectRef[Block]],
        metadata: List[BlockMetadata],
        path: str,
        dataset_uuid: str,
        filesystem: Optional["pyarrow.fs.FileSystem"] = None,
        try_create_dir: bool = True,
        open_stream_args: Optional[Dict[str, Any]] = None,
        block_path_provider: BlockWritePathProvider = DefaultBlockWritePathProvider(),
        write_args_fn: Callable[[], Dict[str, Any]] = lambda: {},
        _block_udf: Optional[Callable[[Block], Block]] = None,
        **write_args,
    ) -> List[ObjectRef[WriteResult]]:
        """Creates and returns write tasks for a file-based datasource."""
        path, filesystem = _resolve_paths_and_filesystem(path, filesystem)
        path = path[0]
        if try_create_dir:
            filesystem.create_dir(path, recursive=True)
        filesystem = _wrap_s3_serialization_workaround(filesystem)

        _write_block_to_file = self._write_block

        if open_stream_args is None:
            open_stream_args = {}

        def write_block(write_path: str, block: Block):
            logger.debug(f"Writing {write_path} file.")
            fs = filesystem
            if isinstance(fs, _S3FileSystemWrapper):
                fs = fs.unwrap()
            if _block_udf is not None:
                block = _block_udf(block)

            with fs.open_output_stream(write_path, **open_stream_args) as f:
                _write_block_to_file(
                    f,
                    BlockAccessor.for_block(block),
                    writer_args_fn=write_args_fn,
                    **write_args,
                )

        write_block = cached_remote_fn(write_block)

        file_format = self._file_format()
        write_tasks = []
        if not block_path_provider:
            block_path_provider = DefaultBlockWritePathProvider()
        for block_idx, block in enumerate(blocks):
            write_path = block_path_provider(
                path,
                filesystem=filesystem,
                dataset_uuid=dataset_uuid,
                block=block,
                block_index=block_idx,
                file_format=file_format,
            )
            write_task = write_block.remote(write_path, block)
            write_tasks.append(write_task)

        return write_tasks
Creates and returns write tasks for a file-based datasource.
Creates and returns write tasks for a file-based datasource.
Creates and returns write tasks for a file - based datasource .
def do_write(
        self,
        blocks: List[ObjectRef[Block]],
        metadata: List[BlockMetadata],
        path: str,
        dataset_uuid: str,
        filesystem: Optional["pyarrow.fs.FileSystem"] = None,
        try_create_dir: bool = True,
        open_stream_args: Optional[Dict[str, Any]] = None,
        block_path_provider: BlockWritePathProvider = DefaultBlockWritePathProvider(),
        write_args_fn: Callable[[], Dict[str, Any]] = lambda: {},
        _block_udf: Optional[Callable[[Block], Block]] = None,
        **write_args,
    ) -> List[ObjectRef[WriteResult]]:
        path, filesystem = _resolve_paths_and_filesystem(path, filesystem)
        path = path[0]
        if try_create_dir:
            filesystem.create_dir(path, recursive=True)
        filesystem = _wrap_s3_serialization_workaround(filesystem)
        _write_block_to_file = self._write_block
        if open_stream_args is None:
            open_stream_args = {}
        def write_block(write_path: str, block: Block):
            logger.debug(f"Writing {write_path} file.")
            fs = filesystem
            if isinstance(fs, _S3FileSystemWrapper):
                fs = fs.unwrap()
            if _block_udf is not None:
                block = _block_udf(block)
            with fs.open_output_stream(write_path, **open_stream_args) as f:
                _write_block_to_file(
                    f,
                    BlockAccessor.for_block(block),
                    writer_args_fn=write_args_fn,
                    **write_args,
                )
        write_block = cached_remote_fn(write_block)
        file_format = self._file_format()
        write_tasks = []
        if not block_path_provider:
            block_path_provider = DefaultBlockWritePathProvider()
        for block_idx, block in enumerate(blocks):
            write_path = block_path_provider(
                path,
                filesystem=filesystem,
                dataset_uuid=dataset_uuid,
                block=block,
                block_index=block_idx,
                file_format=file_format,
            )
            write_task = write_block.remote(write_path, block)
            write_tasks.append(write_task)
        return write_tasks
def do_write ( self , blocks : List [ ObjectRef [ Block ] ] , metadata : List [ BlockMetadata ] , path : str , dataset_uuid : str , filesystem : Optional [ "pyarrow.fs.FileSystem" ] = None , try_create_dir : bool = True , open_stream_args : Optional [ Dict [ str , Any ] ] = None , block_path_provider : BlockWritePathProvider = DefaultBlockWritePathProvider ( ) , write_args_fn : Callable [ [ ] , Dict [ str , Any ] ] = lambda : { } , _block_udf : Optional [ Callable [ [ Block ] , Block ] ] = None , ** write_args , ) -> List [ ObjectRef [ WriteResult ] ] : path , filesystem = _resolve_paths_and_filesystem ( path , filesystem ) path = path [ 0 ] if try_create_dir : filesystem . create_dir ( path , recursive = True ) filesystem = _wrap_s3_serialization_workaround ( filesystem ) _write_block_to_file = self . _write_block if open_stream_args is None : open_stream_args = { } def write_block ( write_path : str , block : Block ) : logger . debug ( f"Writing {write_path} file." ) fs = filesystem if isinstance ( fs , _S3FileSystemWrapper ) : fs = fs . unwrap ( ) if _block_udf is not None : block = _block_udf ( block ) with fs . open_output_stream ( write_path , ** open_stream_args ) as f : _write_block_to_file ( f , BlockAccessor . for_block ( block ) , writer_args_fn = write_args_fn , ** write_args , ) write_block = cached_remote_fn ( write_block ) file_format = self . _file_format ( ) write_tasks = [ ] if not block_path_provider : block_path_provider = DefaultBlockWritePathProvider ( ) for block_idx , block in enumerate ( blocks ) : write_path = block_path_provider ( path , filesystem = filesystem , dataset_uuid = dataset_uuid , block = block , block_index = block_idx , file_format = file_format , ) write_task = write_block . remote ( write_path , block ) write_tasks . append ( write_task ) return write_tasks
Creates and returns write tasks for a file-based datasource.
Creates and returns write tasks for a file - based datasource .
"""Creates and returns write tasks for a file-based datasource."""
self
blocks
List[ObjectRef[Block]]
metadata
List[BlockMetadata]
path
str
dataset_uuid
str
filesystem
Optional["pyarrow.fs.FileSystem"]
try_create_dir
bool
open_stream_args
Optional[Dict[str, Any]]
block_path_provider
BlockWritePathProvider
write_args_fn
Callable[[], Dict[str, Any]]
_block_udf
Optional[Callable[[Block], Block]]