U /ey@sddlZddlZddlZddlmZmZmZddlm Z ddl m Z ddl m Z ddlmZGdddZd d Zd d Zd"ddZd#ddZddZddZddZddZd$ddZddZd d!ZdS)%N)expand_paths_if_neededget_fs_token_pathsstringify_path)AbstractFileSystem)config) _is_local_fs)natural_sort_keyc@sxeZdZdZeddZedddZeddd Zedd d Zed d Z edddZ eddZ eddZ dS)Enginez8The API necessary to provide a new Parquet reader/writercs|dkr|ddnd|kr&td|dkrRt|d|d\}}|||fSttsjtd|r|td |t|tttfr|std d d |D}n t|g}t |dd d}fdd |D||fSdS)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.)Nr rb)modestorage_optionsz4Expected fsspec.AbstractFileSystem or 'fsspec'. Got zUCannot specify storage_options when an explicit filesystem object is specified. Got: zempty urlpath sequencecSsg|] }t|qS)r.0urrC/tmp/pip-unpacked-wheel-dbjnr7gq/dask/dataframe/io/parquet/utils.py ]sz-Engine.extract_filesystem..csg|]}|qSr)Z_strip_protocolrr rrrds) pop ValueErrorr isinstancerlisttuplesetrr)clsZurlpath filesystemdataset_optionsopen_file_optionsr_pathsrrrextract_filesystemsB*    zEngine.extract_filesystemNFc Ks tdS)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. NNotImplementedError) rr r" categoriesindexuse_nullable_dtypesgather_statisticsfilterskwargsrrr read_metadatais@zEngine.read_metadatacKs tdS)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. **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 Nr$)rr Zpiececolumnsr'r(r+rrrread_partitionszEngine.read_partitionc KstdS)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 Nr$) rdfr pathappend partition_onZignore_divisionsZ division_infor+rrrinitialize_writes(zEngine.initialize_writecKstdS)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 Nr$)rr/r0r filenamer2Zreturn_metadatar+rrrwrite_partitionszEngine.write_partitioncKs tdS)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`) Nr$)rpartsmetar r0r1r+rrrwrite_metadataszEngine.write_metadatacCs tdS)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. Nr$)rr0r file_pathrrrcollect_file_metadata0szEngine.collect_file_metadatacCs tdS)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. Nr$)rZ meta_listr Zout_pathrrraggregate_metadataDszEngine.aggregate_metadata)NNFNN)F)FNFN)F) __name__ __module__ __qualname____doc__ classmethodr#r,r.r3r5r8r:r;rrrrr s6 X A  )   r c sdd|dD}td}dd|dD}g}|D](\}}|rP||rPd}|||fq6dd|D}|d d dig}d d|D}|s|rt|d trg}t|}t|fd d|D} ndd|D} t|} || | |fS)aPGet the set of names from the pandas metadata section Parameters ---------- pandas_metadata : dict Should conform to the pandas parquet metadata spec Returns ------- index_names : list List of strings indicating the actual index names column_names : list List of strings indicating the actual column names storage_name_mapping : dict Pairs of storage names (e.g. the field names for PyArrow) and actual names. The storage and field names will differ for index names for certain writers (pyarrow > 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 cSs"g|]}t|tr|dn|qSname)rdict)rnrrrr{sz*_parse_pandas_metadata..Z index_columnsz__index_level_\d+__cSs$g|]}|d|d|dfqS) field_namerB)getrxrrrrsr-NcSsg|]\}}||kr|qSrrr storage_namerBrrrrsZcolumn_indexesrBcSsg|] }|dqSrArrGrrrrsrcsg|]\}}|kr|qSrrrIZindex_storage_names2rrrscSsg|]\}}||kr|qSrrrIrrrrs) recompilematchr1rFrrCrr) Zpandas_metadataZindex_storage_namesZindex_name_xprpairsZpairs2rJZ real_name index_namesZcolumn_index_names column_namesZstorage_name_mappingrrKr_parse_pandas_metadata^s4   rRcs|dk }|dk }|dkr"t|}nt|tr4|g}nt|}|dkrJ|}n0|dkr`g}||}nt|trr|g}nt|}|r|s|fdd|DnP|r|s|fdd|Dn0|r|r||trtdn||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 NFcsg|]}|kr|qSrrrG)rPrrrsz,_normalize_index_columns..csg|]}|kr|qSrrrG)rQrrrsz3Specified index and column names must not intersect)rrstrr intersectionr)Z user_columnsZ data_columnsZ user_indexZ data_indexZspecified_columnsZspecified_indexr)rQrPr_normalize_index_columnss:     rUFcCs(t|td}t|||d\}}|||fS)N)key)root)sortedr_analyze_paths) file_listr rWbasefnsrrr_sort_and_analyze_pathss r]c sfddfdd|D}|dkr|ddd|D]H}t|d }tt|D]\}\}}||krX|}qvqXd|q:tn6|d ttfd d |Dstd g} |D]}| d |dqd | 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 sbfdd}d}|r|dr|dddkrNd}t|}|ddd|d<nLjdkr|ddddr|ddd }t|}|dd d|d<g}t|D]\}}||||dqg}|D]h}|d krq|d kr,|r|d d kr||n|n|r td n ||q||q|sP|rJ|}nd }n|d |}|S)NcsV|jd}|dkrdS|ddkr2|dd}|dkrR|ddkrR|dd}|S)N/.rr)replacesep)iprrr_scrubs   z2_analyze_paths.._join_path.._scrubr_rr^r\z:/r`z..razcan not get parent of root) rrc startswith enumerateextendsplitr1r Exceptionjoin) r0rfZ abs_prefixZ _scrubbedrdreZsimplersZjoinedrrr _join_pathsB         z"_analyze_paths.._join_pathcsg|]}|dqS)r^)rl)rfn)rprrr2sz"_analyze_paths..FrNrarr^c3s|]}|dkVqdS)Nr)rre)basepathlrr @sz!_analyze_paths..z(All paths must begin with the given root)lenrjziprlallAssertionErrorr1rn) rZr rWZpath_parts_list path_partsjkZ base_partZ path_partZout_listr)rprrr rsrrYs6 6   rYc CsDt|dkriSt|dkr4t|dks,t|dSt|dkrtt|}||d|d|dgd}n ||ddd|ddgd}d}t|dkrt|}t|D]\}}|d} |dkr|d||d| |d| dd q|d||jdd| f |jdd| df d q|SdS) z~Utility to aggregate the statistics for N row-groups into a single dictionary. Used by `Engine._construct_parts` rrnum-rowstotal_byte_size)Z file_path_0r|znum-row-groupsr}r-Nr-)rBminmax) rurxpdZ DataFramesumcountrjr1Zilocrr) r9file_row_group_statsfile_row_group_column_statsstat_col_indicesZdf_rgsroZdf_colsindrBrdrrr_aggregate_statsOsL              rc Csng} g} |rt|}d} |D]\} } t| }| rNdgtt| ||}ntt| ||}|D]}||}|dkr| r|dkr| }d} ||}|dkr|} | ||}|| |f|}|dkrqb| ||rbt| || |||| |||}| |qbq"n`|D]V\} } || | f|}|dkr2q| ||rt| || || |}| |q| | fS)NrT)intitemsrurranger1r)r)split_row_groupsaggregation_depthZfile_row_groupsrrrZmake_part_funcZmake_part_kwargsr6statsZresidualr4 row_groupsZrow_group_countZ_rgsrdZi_endZ _residualZrg_listpartstatrrr_row_groups_to_partssp      rcCs<|}t|tr8||kr*t|||}nt|d|S)Nz) is not a recognized directory partition.)rrSrur'r)Zaggregate_filesZpartition_namesrrrr_get_aggregation_depths  rcCs,|dkr(dt|rdnd}t|dS|S)Nz%dataframe.parquet.metadata-task-size-localremoter)rrrF)Zmetadata_task_sizer Z config_strrrr_set_metadata_task_sizes  r readaheadTc Cs|pi}|di}|s$i}d|kr|dddkrj|dd|d<|||||d|dn |d||d<|d d |d <||fS) Nprecache_optionsZopen_file_funcmethodZparquetZ cache_typer6engine)metadatar-rrr r )copyrrFupdate)r rr-rZdefault_engineZ default_cacheZallow_precacherrrr_process_open_file_optionss.   rcKsh|}d|krtdt|di|di}|di}|di}||||fS)Nfilez^Passing user options with the 'file' argument is now deprecated. Please use 'dataset' instead.Zdatasetreadr )rwarningswarn FutureWarningr)r+Z user_kwargsrZ read_optionsr rrr_split_user_options5s rcCs4|st|dkr|s||r$d}n|s,d}t|S)NrTF)rrTbool)r) chunksizerrZfilter_columnsZ stat_columnsrrr_set_gather_statisticsPs  r)F)F)NNNNrT)rLrZpandasrZ fsspec.corerrrZ fsspec.specrZdaskrZdask.dataframe.io.utilsrZ dask.utilsrr rRrUr]rYrrrrrrrrrrrs6    ST>  Y@Q! &