o $ie,@s:ddlZddlZddlmZmZmZmZmZddlm Z ddl m Z ddl m Z ddlmZddlmZddlmZdd lmZdd lmZmZdd lmZmZdd lmZdd lmZm Z ddl!m"Z"m#Z#ddl$m%Z%ddl&m'Z'erxddl(Z(e)e*Z+GdddedZ,e'Gddde,Z-e'Gddde,Z.dS)N) TYPE_CHECKINGAnyDictIterableOptional)urlparse)call_with_retry)%add_creatable_buckets_param_if_s3_uri)DelegatingBlockBuilder) TaskContext)WRITE_UUID_KWARG_NAME)SaveMode)RetryingPyFileSystem_is_local_scheme)Block BlockAccessor) DataContext)Datasink WriteResult)FilenameProvider_DefaultFilenameProvider)_resolve_paths_and_filesystem) DeveloperAPIc@seZdZddddddejddededdedeeee fd ee d eed eed efd dZ deddfddZ d&dedddfddZ defddZdeededdfddZdededefdd Zd!edfd"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|_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_pathrrrwrapretried_io_errorslenr!rrrrrr has_created_dir _skip_write_write_started) selfr!rrrrrrr pathsr/^/home/ubuntu/veenaModal/venv/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/r0r5Wsz _FileDatasink.open_output_streamschemazpyarrow.SchemacCs|jrdSd|_ddlm}|j|jj|ju}|r\|jt j kr*t d|jd|jt j krBt d|jd|jd|_dS|jt jkr\t d|jd|j|j|j||j|_dS) NTrFileTypezPath zO already exists. If this is unexpected, use mode='ignore' to ignore those filesz [SaveMode=z ] Skipping z] Replacing contents )r, pyarrow.fsr8r get_file_infor!typeNotFoundr r ERROR ValueErrorIGNOREloggerwarningr+ OVERWRITEdelete_dir_contents _create_dirr*)r-r6r8 dir_existsr/r/r0on_write_startZs&     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. rr7s3T) recursiveF) r9r8rschemer%s3_try_create_dirrrr:r;r<r create_dir)r-destr8 parsed_uri is_s3_uriskip_create_dir_for_s3tmpr/r/r0rDts   z_FileDatasink._create_dirblocksctxcCs`t}|D]}||q|}t|}|dkr'td|jdS| |d|dS)NrzSkipped writing empty block to ) r add_blockbuildr for_blocknum_rowsr@rAr! write_block)r-rQrRbuilderblockblock_accessorr/r/r0writes   z_FileDatasink.writerY block_indexcCtr4NotImplementedError)r-rYr\rRr/r/r0rWsz_FileDatasink.write_block write_resultcCs*|jr|jdkr|j|jdSdSdS)Nr)r*rVr delete_dirr!)r-r`r/r/r0on_write_completesz_FileDatasink.on_write_completecCs t|j Sr4)rr&r-r/r/r0supports_distributed_writess z)_FileDatasink.supports_distributed_writesr4)__name__ __module__ __qualname__r APPENDstrrboolrrrr1r5rFrDrrr r[rintrWrrbpropertyrdr/r/r/r0r sN  6  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()) rowfiler3cCr])zWrite a row to a file. Args: row: The row to write. file: The file to write the row to. r^)r-rnror/r/r0write_row_to_filez&RowBasedFileDatasink.write_row_to_filerYr\rRcst|jddD]7\}j|jt|j||}tj |t ddfdd}t |ddj jd qdS) NF)public_row_formatWriting  file.c<}|WddS1swYdSr4)r5rprornr- write_pathr/r0write_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-rYr\rR row_indexfilenameryr/rwr0rWs" z RowBasedFileDatasink.write_blockN) rerfrg__doc__rrirrprrkr rWr/r/r/r0rms rmcsneZdZdZdddeeffddZdedd 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|||_dSr4)superr1_min_rows_per_file)r-r!rfile_datasink_kwargs __class__r/r0r1s zBlockBasedFileDatasink.__init__rYror3cCr])zWrite a block of data to a file. Args: block: The block to write. file: The file to write the block to. r^)r-rYror/r/r0write_block_to_file rqz*BlockBasedFileDatasink.write_block_to_filer\rRcshj|jt|j|}tj|fdd}t ddt |ddj j ddS)Ncrur4)r5rrvrYr-rxr/r0write_block_to_pathrzz?BlockBasedFileDatasink.write_block..write_block_to_pathrsrtr{r|r}) rget_filename_for_blockrr rrrr!r@rrr%r()r-rYr\rRrrr/rr0rWs  z"BlockBasedFileDatasink.write_blockr2cCs|jSr4)rrcr/r/r0min_rows_per_write$sz)BlockBasedFileDatasink.min_rows_per_write)rerfrgrrrkr1rrr rWrlr __classcell__r/r/rr0rs r)/loggingrtypingrrrrr urllib.parserray._common.retryrray._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.utilrrray.data.blockrrray.data.contextrray.data.datasource.datasinkrr%ray.data.datasource.filename_providerrrray.data.datasource.path_utilrray.util.annotationsrpyarrow getLoggerrer@rrmrr/r/r/r0s4           =