bihddlmZddlZddlZddlZddlmZm Z m Z ddl m Z ddl mZddlmZddlmZmZGdd Zd Zdd Zdd Zd ZdZdZdZ ddZdZdZddZy)) annotationsN)expand_paths_if_neededget_fs_token_pathsstringify_path)AbstractFileSystem)config) _is_local_fs)natural_sort_key parse_bytesceZdZdZedZedZe d dZe d dZedZ ed dZ ed Z ed Z y) Enginez8The API necessary to provide a new Parquet reader/writerc||jdd}nd|vr td|}|dvrt|d|\}}}||||fSt|tstd||rtd |t|t t tfr&|s td |D cgc] } t| }} n t|g}t|dd |d}||D cgc]} |j| c} ||fScc} wcc} w) aExtract filesystem object from urlpath or user arguments This classmethod should only be overridden for engines that need to handle filesystem implementations other than ``fsspec`` (e.g. ``pyarrow.fs.S3FileSystem``). Parameters ---------- urlpath: str or List[str] Source directory for data, or path(s) to individual parquet files. filesystem: "fsspec" or fsspec.AbstractFileSystem Filesystem backend to use. Default is "fsspec" dataset_options: dict Engine-specific dataset options. open_file_options: dict Options to be used for file-opening at read time. storage_options: dict Options to be passed on to the file-system backend. Returns ------- fs: Any A global filesystem object to be used for metadata processing and file-opening by the engine. paths: List[str] List of data-source paths. dataset_options: dict Engine-specific dataset options. open_file_options: dict Options to be used for file-opening at read time. NfsfsspeczPCannot specify a filesystem argument if the 'fs' dataset option is also defined.)Nrrb)modestorage_optionsz4Expected fsspec.AbstractFileSystem or 'fsspec'. Got zUCannot specify storage_options when an explicit filesystem object is specified. Got: zempty urlpath sequence) pop ValueErrorr isinstancerlisttuplesetrr_strip_protocol) clsurlpath filesystemdataset_optionsopen_file_optionsrr_pathsus Z/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/dask/dataframe/io/parquet/utils.pyextract_filesystemzEngine.extract_filesystemsHT   $$T84B& ;B ! !-dOLB5uo/@@ @b"45 J2$O!<>6=>>!,>>)'23*7D!RFE0561##A&6!   ?7s C0C5cy)Nz256 MiB)rs r$default_blocksizezEngine.default_blocksizeksc t)aRead a single piece of a Parquet dataset into a Pandas DataFrame This function is called many times in individual tasks Parameters ---------- fs: FileSystem piece: object This is some token that is returned by Engine.read_metadata. Typically it represents a row group in a Parquet dataset columns: List[str] List of column names to pull out of that row group index: str, List[str], or False The index name(s). use_nullable_dtypes: boolean Whether to use pandas nullable dtypes (like "string" or "Int64") where appropriate when reading parquet files. dtype_backend: {"numpy_nullable", "pyarrow"} Whether to use pandas nullable dtypes (like "string" or "Int64") where appropriate when reading parquet files. convert_string: boolean Whether to use pyarrow strings when reading parquet files. **kwargs: Includes `"kwargs"` values stored within the `parts` output of `engine.read_metadata`. May also include arguments to be passed to the backend (if stored under a top-level `"read"` key). Returns ------- A Pandas DataFrame NotImplementedError)rrpiececolumnsindexuse_nullable_dtypeskwargss r$read_partitionzEngine.read_partitionosF"##r)Nc t)a#Perform engine-specific initialization steps for this dataset Parameters ---------- df: dask.dataframe.DataFrame fs: FileSystem path: str Destination directory for data. Prepend with protocol like ``s3://`` or ``hdfs://`` for remote data. append: bool If True, may use existing metadata (if any) and perform checks against the new data being stored. partition_on: List(str) Column(s) to use for dataset partitioning in parquet. ignore_divisions: bool Whether or not to ignore old divisions when appending. Otherwise, overlapping divisions will lead to an error being raised. division_info: dict Dictionary containing the divisions and corresponding column name. **kwargs: dict Other keyword arguments (including `index_cols`) Returns ------- tuple: engine-specific instance list of filenames, one per partition r+) rdfrpathappend partition_onignore_divisions division_infor1s r$initialize_writezEngine.initialize_writes P"!r)c t)a  Output a partition of a dask.DataFrame. This will correspond to one output file, unless partition_on is set, in which case, it will correspond to up to one file in each sub-directory. Parameters ---------- df: dask.dataframe.DataFrame path: str Destination directory for data. Prepend with protocol like ``s3://`` or ``hdfs://`` for remote data. fs: FileSystem filename: str partition_on: List(str) Column(s) to use for dataset partitioning in parquet. return_metadata : bool Whether to return list of instances from this write, one for each output file. These will be passed to write_metadata if an output metadata file is requested. **kwargs: dict Other keyword arguments (including `fmd` and `index_cols`) Returns ------- List of metadata-containing instances (if `return_metadata` is `True`) or empty list r+)rr4r5rfilenamer7return_metadatar1s r$write_partitionzEngine.write_partitions >"!r)c t)a; Write the shared metadata file for a parquet dataset. Parameters ---------- parts: List Contains metadata objects to write, of the type undrestood by the specific implementation meta: non-chunk metadata Details that do not depend on the specifics of each chunk write, typically the schema and pandas metadata, in a format the writer can use. fs: FileSystem path: str Output file to write to, usually ``"_metadata"`` in the root of the output dataset append: boolean Whether or not to consolidate new metadata with existing (True) or start from scratch (False) **kwargs: dict Other keyword arguments (including `compression`) r+)rpartsmetarr5r6r1s r$write_metadatazEngine.write_metadatas 0"##r)ct)a Collect parquet metadata from a file and set the file_path. Parameters ---------- path: str Parquet-file path to extract metadata from. fs: FileSystem file_path: str Relative path to set as `file_path` in the metadata. Returns ------- A metadata object. The specific type should be recognized by the aggregate_metadata method. r+)rr5r file_paths r$collect_file_metadatazEngine.collect_file_metadatas $"##r)ct)a Aggregate a list of metadata objects and optionally write out the final result as a _metadata file. Parameters ---------- meta_list: list List of metadata objects to be aggregated into a single metadata object, and optionally written to disk. The specific element type can be engine specific. fs: FileSystem out_path: str or None Directory to write the final _metadata file. If None is specified, the aggregated metadata will be returned, and nothing will be written to disk. Returns ------- If out_path is None, an aggregate metadata object is returned. Otherwise, None is returned. r+)r meta_listrout_paths r$aggregate_metadatazEngine.aggregate_metadata s ."##r)F)FNFN) __name__ __module__ __qualname____doc__ classmethodr%r(r2r:r>rBrErIr'r)r$r r sBVVp}t|dz }tt |D]\}\} } | | k7s|}n d| @t  n>||jd t  t fd|DsJd g} |D]%}| j dj | d'dj | fScc}w) zConsolidate list of file-paths into parquet relative paths Note: This function was mostly copied from dask/fastparquet to use in ArrowEngine`.c fd}d}|rm|drh|dddk(rd}t|}|ddd|d<nD jdk(r5|dddjdr|ddd}t|}|ddd|d<g}t|D],\}}|j |||j d.g}|D]j}|d k(r |d k(rL|r+|d d k(r|j |*|j;|r td |j |Z|j |l|s |r|}|Sd }|S|dj|z}|S) Nc|jjd}|dk(ry|ddk(r|dd}|dkDr |ddk(r|dd}|S)N/.rr)replacesep)iprs r$_scrubz2_analyze_paths.._join_path.._scrubrsZ "&&#&ABwu|crF1u1abEHr)rjrrir\z:/rkz..rlzcan not get parent of root) rrn startswith enumerateextendsplitr6r Exceptionjoin) r5rq abs_prefix _scrubbedrorpsimplersjoinedrs r$ _join_pathz"_analyze_paths.._join_pathqs}  DGAwqzS  Dzq'!"+Q4DGABK$:$:4$@!!WQq\ Dzq'!"+Q dO 6DAq   VAq\//4 5 6 "ACxdr{d*q) #$@AANN1%q! " #    388G#45F r)riFrNrlrc3.K|] }|dk(yw)Nr').0rpbasepathls r$ z!_analyze_paths..s! "#AbqEX  sz(All paths must begin with the given root)rwlenruzipallr6ry)rbrr_rfnpath_parts_list path_partsjk base_part path_partout_listrrs ` @@r$raraksJ 4l?G"#34 ICaA4Q7:4Q7A>8;AEB V# iL''z(BCiL''$(#)#)*4 !ad+22488: aQh/668<<>$\\!QU(3779 V# iL''z(BCiL''$(#)#)*4 3 Br)c g} g} |rt|}d} |jD]\} } t| }| rdgtt | ||z}ntt | ||}|D]s}||z}|dur| r |dk(r| }d} ||z }|dkDr|} | ||}|| |fi|}|5| j ||sIt | || |||| |||}| j |u| | fS|jD]K\} } || | fi|}|| j ||s't | || || |}| j |M| | fS)NrT)intitemsrrranger6r)gather_statisticssplit_row_groupsaggregation_depthfile_row_groupsrrrmake_part_funcmake_part_kwargsr@statsresidualr< row_groupsrow_group_count_rgsroi_end _residualrg_listpartstats r$_row_groups_to_partsrs E E/0$3$9$9$;# ' Hj!*oOsT%/CS"TUUE(O=MNO ',,$,AF (#$ % 7I 1}#,$Qu-%' < T"$+ ,X6q?3H=aF( D LL&9 '# 'p %<'%4$9$9$; # Hj!#D | LL  '(2/9$   T"# #& %<r)c|}t|tr1||vrt||j|z }|St |d|S)Nz) is not a recognized directory partition.)rrQrr/r)aggregate_filespartition_namesrs r$_get_aggregation_depthrasi (/3' o -!$O 47L7L8!   "##LM  r)cZ|(dt|rdndz}tj|dS|S)Nz%dataframe.parquet.metadata-task-size-localremoter)r rget)metadata_task_sizer config_strs r$_set_metadata_task_sizers;!=#B'GX zz*a(( r)c ||xsij}|jdij}|si}d|vr}|jddk(r?|jdd|d<|j||||jd|d||fS|jd||d<|jd d |d <||fS) Nprecache_optionsopen_file_funcmethodparquet cache_typer@engine)metadatar.rrrr)copyrrupdate)r rr.rdefault_engine default_cacheallow_precachers r$_process_open_file_optionsrs+0b668(,,-?DIIK 00    )Y 6.?.C.Cg/ l +  # # (&",.228^L   . .. /@.C.Cm/ l +):(=(=fd(K f % . ..r)c n|j}d|vrtjdti|j dij|j dij}|j dij}|j dij}||||fS)Nfilez^Passing user options with the 'file' argument is now deprecated. Please use 'dataset' instead.datasetreadr )rwarningswarn FutureWarningr)r1 user_kwargsr read_optionsr s r$_split_user_optionsrs++-K   -   //&" % * * , //)R ( - - /O??62.335L#(;R@EEG  r)c|r|dus!t|dkDr|s|j|r d}t|S|sd}t|S)NTrF)rrRbool)r blocksizerrfilter_columns stat_columnss r$_set_gather_statisticsrs] '4/  !A %*;  & &| 4 ! ! "" " ! ""r)cZ|r)t|}|stj|d|zkDryy)NradaptiveF)r npr)row_group_sizesrrs r$_infer_split_row_groupsrs. * bff_5I E r)rJ)NNNN readaheadT) __future__rrnumpyrpandasr fsspec.corerrr fsspec.specrdaskrdask.dataframe.io.utilsr dask.utilsr r r r\rerarrrrrrrrr'r)r$rs"RR*04U$U$p;%| VrL^K\B   #/L6#Br)