o 1 i"+@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$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)Zretryable_errorsrF)rrZ get_current _data_contextunresolved_pathrrrwrapretried_io_errorslenr!rrrrrr has_created_dir) selfr!rrrrrrr pathsr+m/home/app/PaddleOCR-VL-test/.venv_paddleocr/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,r1Usz _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.fsr3r get_file_infor!typeNotFoundr r ERROR ValueErrorZIGNOREloggerwarningZ OVERWRITEZdelete_dir_contents _create_dirr()r)r3Z 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. rr2Zs3T) recursiveF) r4r3rschemer#Zs3_try_create_dirrrr5r6r7r Z create_dir)r)destr3Z parsed_uriZ is_s3_uriZskip_create_dir_for_s3tmpr+r+r,r<ks   z_FileDatasink._create_dirblocksctxcCs`t}|D]}||q|}t|}|dkr'td|jdS| |d|dS)NrzSkipped writing empty block to ) r Z add_blockbuildrZ for_blocknum_rowsr:r;r! write_block)r)rBrCbuilderblockZblock_accessorr+r+r,writes   z_FileDatasink.writerH block_indexcCtr0NotImplementedError)r)rHrJrCr+r+r,rFsz_FileDatasink.write_block write_resultcCs*|jr|jdkr|j|jdSdSdS)Nr)r(rErZ delete_dirr!)r)rNr+r+r,on_write_completesz_FileDatasink.on_write_completecCs t|j Sr0)rr$r)r+r+r,supports_distributed_writess z)_FileDatasink.supports_distributed_writes)r.N)__name__ __module__ __qualname__r APPENDstrrboolrrrr-r1r=r<rrr rIrintrFrrOpropertyrQr+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()) rowfiler/cCrK)zWrite a row to a file. Args: row: The row to write. file: The file to write the row to. rL)r)r[r\r+r+r,write_row_to_filez&RowBasedFileDatasink.write_row_to_filerHrJrCcst|jddD]7\}j|jt|j||}tj |t ddfdd}t |ddj jd qdS) NF)Zpublic_row_formatWriting  file.c<}|WddS1swYdSr0)r1r]r\r[r) write_pathr+r,write_row_to_path "z;RowBasedFileDatasink.write_block..write_row_to_pathwrite '' descriptionmatch) enumerateZ iter_rowsrZget_filename_for_rowkwargsr task_idx posixpathjoinr!r:debugrr#r&)r)rHrJrCZ row_indexfilenamerer+rcr,rFs" z RowBasedFileDatasink.write_blockN) rRrSrT__doc__rrVrr]rrXr rFr+r+r+r,rZs rZcsneZdZdZdddeeffddZdedd 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_fileruc stj|fi|||_dSr0)superr-_min_rows_per_file)r)r!ruZfile_datasink_kwargs __class__r+r,r-s zBlockBasedFileDatasink.__init__rHr\r/cCrK)zWrite a block of data to a file. Args: block: The block to write. file: The file to write the block to. rL)r)rHr\r+r+r,write_block_to_filer^z*BlockBasedFileDatasink.write_block_to_filerJrCcshj|jt|j|}tj|fdd}t ddt |ddj j ddS)Ncrar0)r1rzrbrHr)rdr+r,write_block_to_pathrfz?BlockBasedFileDatasink.write_block..write_block_to_pathr_r`rgrhri) rZget_filename_for_blockrmr rnrorpr!r:rqrr#r&)r)rHrJrCrrr|r+r{r,rF s  z"BlockBasedFileDatasink.write_blockr.cCs|jSr0)rwrPr+r+r,min_rows_per_writesz)BlockBasedFileDatasink.min_rows_per_write)rRrSrTrsrrXr-rrzr rFrYr} __classcell__r+r+rxr,rts rt)/loggingrotypingrrrrr urllib.parserZray._common.retryrZray._private.arrow_utilsr Z+ray.data._internal.delegating_block_builderr Z'ray.data._internal.execution.interfacesr Z(ray.data._internal.planner.plan_write_opr Zray.data._internal.savemoder Zray.data._internal.utilrrZray.data.blockrrZray.data.contextrZray.data.datasource.datasinkrrZ%ray.data.datasource.filename_providerrrZray.data.datasource.path_utilrZray.util.annotationsrZpyarrow getLoggerrRr:rrZrtr+r+r+r,s4            =