>ieddlmZddlZddlZddlZddlZddlm Z m Z m Z ddl m Z ddlmZddlmZddlmZmZGdd Zd Zd Zdd ZddZdZdZdZdZ ddZdZdZ ddZ!dS)) annotationsN)expand_paths_if_neededget_fs_token_pathsstringify_path)AbstractFileSystem)config) _is_local_fs)natural_sort_key parse_bytesceZdZdZedZe d dZedZe ddZe ddZ ed Z edd Z ed Z ed Z dS)Enginez8The API necessary to provide a new Parquet reader/writerc||ddnd|vrtd|dvrt|d|\}}|||fSttstd|rtd |t|t t tfr|std d |D}nt|g}t|dd d}fd |D||fS)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 sequencec,g|]}t|S)r).0us ?lib/python3.11/site-packages/dask/dataframe/io/parquet/utils.py z-Engine.extract_filesystem..`s >>>>!,,>>>c:g|]}|Sr)_strip_protocol)rrrs rrz-Engine.extract_filesystem..gs'6661##A&&666r) pop ValueErrorr isinstancerlisttuplesetrr) clsurlpath filesystemdataset_optionsopen_file_optionsr_pathsrs @rextract_filesystemzEngine.extract_filesystemsT   $$T844BB&& ;B ! ! !-dOLB5uo/@@ @b"455  O2OO !N>>>>g>>>)'223*7D!RFFE6666666!  rNc  t)aGather metadata about a Parquet Dataset to prepare for a read This function is called once in the user's Python session to gather important metadata about the parquet dataset. Parameters ---------- fs: FileSystem paths: List[str] A list of paths to files (or their equivalents) categories: list, dict or None Column(s) containing categorical data. index: str, List[str], or False The column name(s) to be used as the index. If set to ``None``, pandas metadata (if available) can be used to reset the value in this function use_nullable_dtypes: boolean Whether to use pandas nullable dtypes (like "string" or "Int64") where appropriate when reading parquet files. gather_statistics: bool Whether or not to gather statistics to calculate divisions for the output DataFrame collection. filters: list List of filters to apply, like ``[('x', '>', 0), ...]``. **kwargs: dict (of dicts) User-specified arguments to pass on to backend. Top level key can be used by engine to select appropriate dict. Returns ------- meta: pandas.DataFrame An empty DataFrame object to use for metadata. Should have appropriate column names and dtypes but need not have any actual data statistics: Optional[List[Dict]] Either None, if no statistics were found, or a list of dictionaries of statistics data, one dict for every partition (see the next return value). The statistics should look like the following: [ {'num-rows': 1000, 'columns': [ {'name': 'id', 'min': 0, 'max': 100}, {'name': 'x', 'min': 0.0, 'max': 1.0}, ]}, ... ] parts: List[object] A list of objects to be passed to ``Engine.read_partition``. Each object should represent a piece of data (usually a row-group). The type of each object can be anything, as long as the engine's read_partition function knows how to interpret it. NotImplementedError) r$rr* categoriesindexuse_nullable_dtypes dtype_backendgather_statisticsfilterskwargss r read_metadatazEngine.read_metadatalsB"###rcdS)Nz256 MiBr)r$s rdefault_blocksizezEngine.default_blocksizesyrFc 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 r-)r$rpiececolumnsr0r1r5s rread_partitionzEngine.read_partitionsF"###rc 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-) r$dfrpathappend partition_onignore_divisions division_infor5s rinitialize_writezEngine.initialize_writes P"!rc 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-)r$r>r?rfilenamerAreturn_metadatar5s rwrite_partitionzEngine.write_partitions >"!rc 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-)r$partsmetarr?r@r5s rwrite_metadatazEngine.write_metadata#s0"###rct)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-)r$r?r file_paths rcollect_file_metadatazEngine.collect_file_metadata=s$"###rct)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 raggregate_metadatazEngine.aggregate_metadataQs."###r)NNNNNNF)FNFN)__name__ __module__ __qualname____doc__ classmethodr+r6r8r<rDrHrLrOrSrrrr r sPBBVV[Vp  @$@$@$[@$D[ 0.8). column_indexes_names : list The names for ``df.columns.name`` or ``df.columns.names`` for a MultiIndex in the columns Notes ----- This should support metadata written by at least * fastparquet>=0.1.3 * pyarrow>=0.7.0 cLg|]!}t|tr|dn|"Sname)r dict)rns rrz*_parse_pandas_metadata..s?  4((/& ar index_columnsz__index_level_\d+__cVg|]&}|d|d|df'S) field_namer])getrxs rrz*_parse_pandas_metadata..s@   89|QvY ' '63   rr;Nc$g|] \}}||k |Srrr storage_namer]s rrz*_parse_pandas_metadata..s'SSS0\4dl>R>R4>R>R>Rrcolumn_indexesr]cg|] }|d Sr\rrds rrz*_parse_pandas_metadata..s@@@!F)@@@rrc"g|] \}}|v | Srr)rrhr]index_storage_names2s rrz*_parse_pandas_metadata..s/   )lDdBV6V6VD6V6V6Vrc$g|] \}}||k |Srrrgs rrz*_parse_pandas_metadata..s'XXX!5,4|r8|r6||trt dn||fS)aDNormalize user and file-provided column and index names Parameters ---------- user_columns : None, str or list of str data_columns : list of str user_index : None, str, or list of str data_index : list of str Returns ------- column_names : list of str index_names : list of str NFcg|]}|v| Srr)rrerws rrz,_normalize_index_columns..s#HHHa1K3G3G3G3G3Grcg|]}|v| Srr)rrerys rrz,_normalize_index_columns..s#FFFQ0E0Eq0E0E0Erz3Specified index and column names must not intersect)r!r strr# intersectionr) user_columns data_columns user_index data_indexspecified_columnsspecified_indexryrws @@r_normalize_index_columnsrsu%D0 ,OL)) L# & &*$~ L))  u   !L0 J $ $& \ *%% !0!! HHHH<HHH !?!$ FFFF*FFF  !. !$  |   ) )+ 6 6 TRSS S T$   $$rFcbt|t}t|||\}}|||fS)N)key)root)sortedr _analyze_paths) file_listrrbasefnss r_sort_and_analyze_pathsrs:y&6777Iy"4888ID# dC rc\ fd fd|D}|durq|ddd |D]N}t|dz }tt |D]\}\}}||kr|}n d| Ot  nS |d t  t fd |Ds Jd g} |D]2}| d| d3d | fS) zConsolidate list of file-paths into parquet relative paths Note: This function was mostly copied from dask/fastparquet to use in both `FastParquetEngine` and `ArrowEngine`.c@ fd}d}|r|dr|dddkr%d}t|}|ddd|d<n` jdkrU|ddddr2|ddd}t|}|ddd|d<g}t|D]7\}}||||d8g}|D]}|d kr |d kr`|r7|d d kr||3|H|rtd ||o|||s|r|}nd }n|d |z}|S) Nc|jd}|dkrdS|ddkr |dd}|dkr|ddkr |dd}|S)N/.rr)replacesep)iprs r_scrubz2_analyze_paths.._join_path.._scrub si "&#&&ABwwsu||crcF1uu1abbEHrrrrr\z:/rz..rzcan not get parent of root) r!r startswith enumerateextendsplitr@r Exceptionjoin) r?r abs_prefix _scrubbedrrsimplersjoinedrs r _join_pathz"_analyze_paths.._join_path s       &DG &AwqzS   Dzzq'!""+Q4DGABBK$:$:4$@$@!!WQqS\ Dzzq'!""+Q dOO 6 6DAq   VVAq\\//44 5 5 5 5 " "ACxxd&r{d**q)))) &#$@AAANN1%%%%q!!!! 6 #388G#4#45F rcLg|] }|d!S)r)r)rfnrs rrz"_analyze_paths..?s/EEERzz"~~++C00EEErFrNrrrc34K|]}|dkVdS)Nr)rrbasepathls r z!_analyze_paths..MsA  "#AbqbEX       rz(All paths must begin with the given root)lenrziprallr@r) rrrpath_parts_list path_partsjk base_part path_partout_listrrrs ` @@@rrrs 44444lFEEE9EEEO u}}"1%crc*) $ $JJ!#A-6s8Z7P7P-Q-Q  ))Iy ))AE* |HH MM:d##))#.. MM     '6      6 6 5 6 6 6H%   HHZ^ $ $     rct|dkriSt|dkrt|dksJ|dSt|dkretj|}||d|d|dgd}n ||ddd|ddgd}d}t|dkrtj|}t |D]^\}}|dz} |w|d| } |d| dz} |d| dz} | | kr | r|d d | ic|d || | | d |jdd| f } |jdd| dzf } |jdd| dzf} | | kr!| r|d d | i>|d || | | d `|S) z~Utility to aggregate the statistics for N row-groups into a single dictionary. Used by `Engine._construct_parts` rrnum-rowstotal_byte_size) file_path_0rznum-row-groupsrr;Nrr; null_count)r]minmaxr) rpd DataFramesumcountrr@ilocdropnarr) rNfile_row_group_statsfile_row_group_column_statsstat_col_indicesdf_rgsrdf_colsindr]rminvalmaxvalrs r_aggregate_statsr\s   1$$ ( ) )Q . .'((A----#A&& # $ $q ( (\"677F(":.2244"("4":":"<"<#)*;#<#@#@#B#B AA )03J?"##7#:;L#M A * + +a / /l#>??G"#344  ICaA4Q7:4Q7A>8;AEB V## #iL''z(BCCCCiL''$(#)#)*4 !aaad+224488:: aaaQh/6688<<>>$\!!!QU(37799 V## #iL''z(BCCCCiL''$(#)#)*4 rc &g} g} |rt|}d} |D]\} } t| }| r#dgtt | ||z}ntt | ||}|D]}||z}|dur| r |dkr| }d} ||z }|dkr|} | ||}|| |fi|}|;| ||rCt | || |||| |||}| |nq|D]\\} } || | fi|}|| ||r3t | || || |}| |]| | fS)NrT)intitemsrr!ranger@r)r3split_row_groupsaggregation_depthfile_row_groupsrrrmake_part_funcmake_part_kwargsrJstatsresidualrF row_groupsrow_group_count_rgsri_end _residualrg_listpartstats r_row_groups_to_partsrsS E E;#/00$3$9$9$;$;# '# ' Hj!*ooO PsT%/CS"T"TUUUE(O=MNNOO ' ',,$,,%AFF (#$ % 7I 1}}#,$QuW-%~' < T"""$'+ ,X6qw?3H=agF( D LL&&&9 '# 'J%4$9$9$;$; # # Hj!>#D | LL     #'(2/9$   T""" %<rc|}t|tr<||vr&t|||z }nt |d|S)Nz) is not a recognized directory partition.)r rrr0r)aggregate_filespartition_namesrs r_get_aggregation_depthrs| (/3''  o - -!$O 4 47L7L88!  "MMM  rc`|+dt|rdndz}tj|dS|S)Nz%dataframe.parquet.metadata-task-size-localremoter)r rrc)metadata_task_sizer config_strs r_set_metadata_task_sizersD!=#B'' 5GGX z*a((( r readaheadTc |pi}|di}|si}d|vr|dddkrH|dd|d<|||||d|dn2|d||d<|d d |d <||fS) Nprecache_optionsopen_file_funcmethodparquet cache_typerJengine)metadatar;rrrr)copyrrcupdate)r(rr;rdefault_engine default_cacheallow_precachers r_process_open_file_optionsr(s&+0b6688(,,-?DDIIKK 000   $ / /9 < <.?.C.Cg// l +  # # (&",.228^LL     /@.C.Cm// l +):(=(=fd(K(K f % . ..rc |}d|vrtjdti|di|di}|di}|di}||||fS)Nfilez^Passing user options with the 'file' argument is now deprecated. Please use 'dataset' instead.datasetreadr()rwarningswarn FutureWarningr)r5 user_kwargsr' read_optionsr(s r_split_user_optionsrNs++--K   -     //&" % % * * , , //)R ( ( - - / /O??62..3355L#(;R@@EEGG  rc|r|dus*t|dkr|s||rd}n|sd}t|S)NTrF)rrbool)r3 blocksizerrfilter_columns stat_columnss r_set_gather_statisticsrisq "'4//  ! !A % %*; %  & &| 4 4 &! "" ! " ""rcf|r.t|}|stj|d|zkrdSdS)NradaptiveF)r npr)row_group_sizesr rs r_infer_split_row_groupsrsC **  bf_55I EE: 5rrT)NNNNrT)" __future__rrnrnumpyrpandasr fsspec.corerrr fsspec.specrdaskrdask.dataframe.io.utilsr dask.utilsr r r r{rrrrrrrrrrrrrrrs"""""" RRRRRRRRRR******00000044444444X$X$X$X$X$X$X$X$v QOQOQOh;%;%;%|    VVVVrLLL^KKK\B     #/#/#/#/L6###Br