o bi +@s2ddlZddlZddlmZmZmZmZmZddlm Z ddl m Z ddl m Z ddlmZddlmZddlmZdd lmZmZmZdd lmZmZdd lmZdd lmZmZdd l m!Z!m"Z"ddl#m$Z$ddl%m&Z&ertddl'Z'e(e)Z*GdddedZ+e&Gddde+Z,e&Gddde+Z-dS)N) TYPE_CHECKINGAnyDictIterableOptional)urlparse)%add_creatable_buckets_param_if_s3_uri)DelegatingBlockBuilder) TaskContext)WRITE_UUID_KWARG_NAME)SaveMode)RetryingPyFileSystem_is_local_schemecall_with_retry)Block BlockAccessor) DataContext)Datasink WriteResult)FilenameProvider_DefaultFilenameProvider)_resolve_paths_and_filesystem) DeveloperAPIc@seZdZddddddejddededdedeeee fd ee d eed eed efd dZ deddfddZ d$ddZ defddZdeededdfddZdededefddZdedfd d!Zedefd"d#ZdS)% _FileDatasinkNT) filesystemtry_create_diropen_stream_argsfilename_provider dataset_uuid file_formatmodepathrzpyarrow.fs.FileSystemrrrrrr c Cs|duri}|durt||d}t|_||_t||\} |_tj|j|jj d|_t | dks7Jt | | d|_ ||_ ||_ ||_||_||_||_d|_dS)a Initialize this datasink. Args: path: The folder to write files to. filesystem: The filesystem to write files to. If not provided, the filesystem is inferred from the path. try_create_dir: Whether to create the directory to write files to. open_stream_args: Arguments to pass to ``filesystem.open_output_stream``. filename_provider: A :class:`ray.data.datasource.FilenameProvider` that generates filenames for each row or block. dataset_uuid: The UUID of the dataset being written. If specified, it's included in the filename. file_format: The file extension. If specified, files are written with this extension. N)rr)retryable_errorsrF)rr get_current _data_contextunresolved_pathrrr wrapretried_io_errorslenr!rrrrrr has_created_dir) selfr!rrrrrrr pathsr-U/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/datasource/file_datasink.py__init__!s*    z_FileDatasink.__init__returnpyarrow.NativeFilecCs|jj|fi|jSN)ropen_output_streamr)r+r!r-r-r.r3Usz _FileDatasink.open_output_streamcCsddlm}|j|jj|ju}|rQ|jtj kr"t d|jd|jtj kr7t d|jd|jdS|jtjkrQt d|jd|j|j|j||j|_dS)NrFileTypezPath zO already exists. If this is unexpected, use mode='ignore' to ignore those filesz [SaveMode=z ] Skipping z] Replacing contents ) pyarrow.fsr5r get_file_infor!typeNotFoundr r ERROR ValueErrorIGNOREloggerwarning OVERWRITEdelete_dir_contents _create_dirr*)r+r5 dir_existsr-r-r.on_write_startXs     z_FileDatasink.on_write_startcCsjddlm}t|}|jdk}|o|jj }|jr3|s3|j|j |j ur3t |}|jj |dddSdS)zsCreate a directory to write files to. If ``try_create_dir`` is ``False``, this method is a no-op. rr4s3T) recursiveF) r6r5rschemer%s3_try_create_dirrrr7r8r9r create_dir)r+destr5 parsed_uri is_s3_uriskip_create_dir_for_s3tmpr-r-r.rAks   z_FileDatasink._create_dirblocksctxcCs`t}|D]}||q|}t|}|dkr'td|jdS| |d|dS)NrzSkipped writing empty block to ) r add_blockbuildr for_blocknum_rowsr=r>r! write_block)r+rNrObuilderblockblock_accessorr-r-r.writes   z_FileDatasink.writerV block_indexcCtr2NotImplementedError)r+rVrYrOr-r-r.rTsz_FileDatasink.write_block write_resultcCs*|jr|jdkr|j|jdSdSdS)Nr)r*rSr delete_dirr!)r+r]r-r-r.on_write_completesz_FileDatasink.on_write_completecCs t|j Sr2)rr&r+r-r-r.supports_distributed_writess z)_FileDatasink.supports_distributed_writes)r0N)__name__ __module__ __qualname__r APPENDstrrboolrrrr/r3rCrArrr rXrintrTrr_propertyrar-r-r-r.r sN  4   rc@s@eZdZdZdeeefddfddZdede d e fd d Z d S) RowBasedFileDatasinkaA datasink that writes one row to each file. Subclasses must implement ``write_row_to_file`` and call the superclass constructor. Examples: .. testcode:: import io from typing import Any, Dict import pyarrow from PIL import Image from ray.data.datasource import RowBasedFileDatasink class ImageDatasink(RowBasedFileDatasink): def __init__(self, path: str, *, column: str, file_format: str = "png"): super().__init__(path, file_format=file_format) self._file_format = file_format self._column = column def write_row_to_file(self, row: Dict[str, Any], file: "pyarrow.NativeFile"): image = Image.fromarray(row[self._column]) buffer = io.BytesIO() image.save(buffer, format=self._file_format) file.write(buffer.getvalue()) rowfiler1cCrZ)zWrite a row to a file. Args: row: The row to write. file: The file to write the row to. r[)r+rkrlr-r-r.write_row_to_filez&RowBasedFileDatasink.write_row_to_filerVrYrOcst|jddD]7\}j|jt|j||}tj |t ddfdd}t |ddj jd qdS) NF)public_row_formatWriting  file.c<}|WddS1swYdSr2)r3rmrlrkr+ write_pathr-r.write_row_to_path "z;RowBasedFileDatasink.write_block..write_row_to_pathwrite '' descriptionmatch) enumerate iter_rowsrget_filename_for_rowkwargsr task_idx posixpathjoinr!r=debugrr%r()r+rVrYrO row_indexfilenamervr-rtr.rTs" z RowBasedFileDatasink.write_blockN) rbrcrd__doc__rrfrrmrrhr rTr-r-r-r.rjs rjcsneZdZdZdddeeffddZdedd fd d Zded ed e fddZ e deefddZ Z S)BlockBasedFileDatasinka)A datasink that writes multiple rows to each file. Subclasses must implement ``write_block_to_file`` and call the superclass constructor. Examples: .. testcode:: class CSVDatasink(BlockBasedFileDatasink): def __init__(self, path: str): super().__init__(path, file_format="csv") def write_block_to_file(self, block: BlockAccessor, file: "pyarrow.NativeFile"): from pyarrow import csv csv.write_csv(block.to_arrow(), file) N)min_rows_per_filerc stj|fi|||_dSr2)superr/_min_rows_per_file)r+r!rfile_datasink_kwargs __class__r-r.r/s zBlockBasedFileDatasink.__init__rVrlr1cCrZ)zWrite a block of data to a file. Args: block: The block to write. file: The file to write the block to. r[)r+rVrlr-r-r.write_block_to_filernz*BlockBasedFileDatasink.write_block_to_filerYrOcshj|jt|j|}tj|fdd}t ddt |ddj j ddS)Ncrrr2)r3rrsrVr+rur-r.write_block_to_pathrwz?BlockBasedFileDatasink.write_block..write_block_to_pathrprqrxryrz) rget_filename_for_blockrr rrrr!r=rrr%r()r+rVrYrOrrr-rr.rT s  z"BlockBasedFileDatasink.write_blockr0cCs|jSr2)rr`r-r-r.min_rows_per_writesz)BlockBasedFileDatasink.min_rows_per_write)rbrcrdrrrhr/rrr rTrir __classcell__r-r-rr.rs r).loggingrtypingrrrrr urllib.parserray._private.arrow_utilsr+ray.data._internal.delegating_block_builderr 'ray.data._internal.execution.interfacesr (ray.data._internal.planner.plan_write_opr ray.data._internal.savemoder ray.data._internal.utilr rrray.data.blockrrray.data.contextrray.data.datasource.datasinkrr%ray.data.datasource.filename_providerrrray.data.datasource.path_utilrray.util.annotationsrpyarrow getLoggerrbr=rrjrr-r-r-r.s2           =