a S/b @sddlZddlmZddlmZddlmZmZmZddl m Z ddl m Z z ddl Z WneyndZ Yn0ddlZddlZddlZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZm Z m!Z!m"Z"m#Z#ddl$m%Z%ddl&m'Z'ddl(m)Z)ddl*m*Z*ddl+m,Z,m-Z-ddl(m.Z.ddl+m/Z/GdddZ0d?ddZ1ddZ2d@ddZ3d d!Z4d"d#Z5d$d%Z6d&d'Z7e7Z8dAd,d-Z9d.Z:d/d0Z;e;ejd5d6Z>dd7d8d9Z?dBdl(mAZAe@jBeAj@_BdS)CN)Mapping)BytesIO)catch_warnings simplefilterwarn)HighLevelGraph)DataFrameIOLayer)compr)get_fs_token_paths)open) open_files)infer_compression)CategoricalDtypeis_datetime64_any_dtypeis_float_dtypeis_integer_dtypeis_object_dtype)tokenize) read_bytes)flatten)delayed) asciitable parse_bytes) new_dd_object)clear_known_categoriesc@s(eZdZdZddZddZddZdS) CSVFunctionWrapperzg CSV Function-Wrapper Class Reads CSV data from disk to produce a partition (given a key). c Cs:||_||_||_||_||_||_||_||_| |_dSN) full_columnscolumnscolnameheadheaderreaderdtypesenforcekwargs) selfrr r!r"r#r$r%r&r'r)4lib/python3.9/site-packages/dask/dataframe/io/csv.py__init__-s zCSVFunctionWrapper.__init__c s\fddjjDjkr$StjjjjjfddDjj S)zUReturn a new CSVFunctionWrapper object with a sub-column projection. csg|]}|vr|qSr)r).0c)r r)r* Hz6CSVFunctionWrapper.project_columns..csi|]}|j|qSr))r%r,)r(r)r* Rr0z6CSVFunctionWrapper.project_columns..) r"r rrr!r#r$r&r')r(r r))r r(r*project_columnsCs z"CSVFunctionWrapper.project_columnsc Cs|\}}}}|dur6|j|tt|j|jjjf}nd}d}|j}|sxd}|dd| dddurx|dd|s|dd|j } d} |j dur|jrd} n|j } | |d<t |j ||j||j| ||j| } | r| |j S| S)NFTskiprowsr#r skipfooterZusecols)r!sortedlistr"cat categoriesr'copypopgetrr pandas_read_textr$r#r%r&) r(partblockpathis_firstis_lastZ path_info write_headerZ rest_kwargsr Zproject_after_readdfr)r)r*__call__WsJ        zCSVFunctionWrapper.__call__N)__name__ __module__ __qualname____doc__r+r2rDr)r)r)r*r'srTFc  Cst} |r"||s"| || || d|| fi|} |rTt| ||r||r|t| jt|kr|td| j||r|\} }} | |} | j fi| t j tt| | | i} | S)aaConvert a block of bytes to a Pandas DataFrame Parameters ---------- reader : callable ``pd.read_csv`` or ``pd.read_table``. b : bytestring The content to be parsed with ``reader`` header : bytestring An optional header to prepend to ``b`` kwargs : dict A dictionary of keyword arguments to be passed to ``reader`` dtypes : dict dtypes to assign to columns path : tuple A tuple containing path column name, path to file, and an ordered list of paths. See Also -------- dask.dataframe.csv.read_pandas_from_bytes rzColumns do not match)r startswithrstripwriteseek coerce_dtypesr6r ValueErrorindexassignpd Categorical from_codesnpZfulllen)r$br#r'r%r rBr&r?ZbiorCr!pathscoder)r)r*r<s"       r<c Csg}g}g}|jD]}||vr|j|||kr|j|}||}t|rdt|rd||||fqt|rt|r||qz||||||<Wqty}z*||||f|||fWYd}~qd}~00q|rl|rd ddt |dddD} d| } d } nd } d } t |d dd}t gd |} d d dd|D} dj | | | | d}nd}|r|rdnd}d dd|D}dj ||d}nd}|s|rdd}d| t d||g}t|dS)zCoerce dataframe to dtypes safely Operates in place Parameters ---------- df: Pandas DataFrame dtypes: dict like {'x': float} N css"|]\}}d|d|VqdS)z- z Nr))r-r.er)r)r* sz coerce_dtypes..cSs t|dSNrstrxr)r)r*r0zcoerce_dtypes..)keyzAThe following columns also raised exceptions on conversion: %s zf Alternatively, provide `assume_missing=True` to interpret all unspecified integer columns as floats.cSs t|dSr\r]r_r)r)r*rar0)ZColumnZFoundZExpectedz dtype={%s}z , css$|]\}}}|d|dVqdS)z: ''Nr))r-kv_r)r)r*r[sz{table} {exceptions}Usually this is due to dask's dtype inference failing, and *may* be fixed by specifying dtypes manually by adding: {dtype_kw} to the call to `read_csv`/`read_table`.{extra})table exceptionsdtype_kwextraz also  css|]}d|VqdS)z- %sNr)r,r)r)r*r[ r0aThe following columns{also}failed to properly parse as dates: {cols} This is usually due to an invalid value in that column. To diagnose and fix it's recommended to drop these columns from the `parse_dates` keyword, and manually convert them to dates later using `dd.to_datetime`.)alsocolsz %s z=-------------------------------------------------------------z=Mismatched dtypes found in `pd.read_csv`/`pd.read_table`. %s)r r%rrappendrrastype Exceptionjoinr5rformatfilterrN)rCr%Z bad_dtypesZ bad_dateserrorsr.actualZdesiredrZexrirkrhrjZ dtype_msgrmrnZdate_msgZrulemsgr)r)r*rMsh    &      rMc s|j} |jdgdj} ttrBfdd| D} | | } n| } | D] }d| |<qJt|j}tt |}tt |}tt |}|r|\}dd|D}rfdd|D}|j fi|t jtjt|tdt|i}||f}t| rt|| d}g}|pd \}}tt|D]2}||||r>||nd ||||gq"d }|t|| ||||}t|||t|d ||||| || |d d }t||i|ti}t|||dt|dS)aConvert blocks of bytes to a dask.dataframe This accepts a list of lists of values of bytes where each list corresponds to one file, and the value of bytes concatenate to comprise the entire file, in order. Parameters ---------- reader : callable ``pd.read_csv`` or ``pd.read_table``. block_lists : list of lists of delayed values of bytes The lists of bytestrings where each list corresponds to one logical file header : bytestring The header, found at the front of the first file, to be prepended to all blocks head : pd.DataFrame An example Pandas DataFrame to be used for metadata. kwargs : dict Keyword arguments to pass down to ``reader`` path : tuple, optional A tuple containing column name for path and the path_converter if provided Returns ------- A dask.dataframe category)Zincludecs0g|](}t|tr|jdur|qSr) isinstancer;rr8)r-re)specified_dtypesr)r*r/Nsz)text_blocks_to_pandas..cSsg|]}|djqS))r?)r-rVr)r)r*r/er0csg|] }|qSr)r))r-p)path_converterr)r*r/gr0)dtype)rn)NNNz read-csv-T)labelZproduces_tasksrr|)r%Zto_dictZ select_dtypesr rzr differencer6tupler block_maskblock_mask_lastrPrQrRrSrTZzerosrUintsetrrangerorr rrr)r$ block_listsr#r"r'r&r{r? blocksizeurlpathr%Z categoricalsZknown_categoricalsZunknown_categoricalsrer Zblocksr@rAr!rWpartsirnameZlayerZgraphr))r~r{r*text_blocks_to_pandassl&           0 rccs6|D],}|sqdVdd|ddDEdHqdS)z Yields a flat iterable of booleans to mark the zeroth elements of the nested input ``block_lists`` in a flattened output. >>> list(block_mask([[1, 2], [3, 4], [5]])) [True, False, True, False, True] Tcss|] }dVqdSFNr)r-rgr)r)r*r[r0zblock_mask..r|Nr)rr>r)r)r*rs rccs6|D],}|sqdd|ddDEdHdVqdS)z Yields a flat iterable of booleans to mark the last element of the nested input ``block_lists`` in a flattened output. >>> list(block_mask_last([[1, 2], [3, 4], [5]])) [False, True, False, True, True] css|] }dVqdSrr)rr)r)r*r[r0z"block_mask_last..NTr)rr)r)r*rs rcCs"d}t|||}t|tdS)N gA)rmin)Z total_memory cpu_countZ memory_factorrr)r)r*auto_blocksizesrcCsdd}tdur`t,tdttj}t}Wdn1sD0Y|r`|r`t||S|S)Niignore)psutilrrRuntimeWarningZvirtual_memorytotalrr)defaultZmemZcpur)r)r*_infer_block_sizes  & rrinferrc $ Ks|j} |dur$t|dkr$|| d<nd}| r:t| tr:d} d| vsJd| vrZtd| dd D]} | | vr^t| d | q^| d drtd | t| d tr| d }}}nR| d durd}}}n6t| d }t |}t tt t|dt|}t| dt r.t d| t| dtrZ| rZ| d| d}nd}|dkrt|d| dd}t|d}|dkrt}t|trt|}|r|rtd|d}|tvrtd||r|r||kr|dkrtd|}|}t|f||||| d| p*i}| rJ|\}}}| |f}n |\}}d}t|dtt fsp|g}|durt|dr|dd}| dd}| d|durdnd}|durdnd}| dr^g}||D]p}|| d}t|dkr:t|ddkrD||dn ||t||krqnqn||||}|sxdnt|t|d }|dur|||krt||krtdt|tr||7}|durd n |||}| }| d!dz|t!|fd |i|} WnDt"j#j$y^}!z&d"t|!vrHtd#|!WYd}!~!n d}!~!00| r~| | j%vr~td$| | d%i}"|"duri}"|rt|"tr| j%D]2}#t&| |#j'r|#|"vr| |#(t)| |#<qd&d'|D}t*|||| | ||"|||d( S))Nr|lineterminatorrYr?rOZ index_colz7Keywords 'index' and 'index_col' not supported. Use dd.z#(...).set_index('my-index') instead)iteratorZ chunksizez not supported for dd.ZnrowszThe 'nrows' keyword is not supported by `dd.{0}`. To achieve the same behavior, it's recommended to use `dd.{0}(...).head(n=nrows)`r3rr#z)List of header rows not supported for dd.Z convertersrrb)modestorage_optionsrrzWarning %s compression does not support breaking apart files Please ensure that each individual file can fit in memory and use the keyword ``blocksize=None to remove this message`` Setting ``blocksize=None``z#Compression format %s not installedz}Unexpected behavior can result from passing skiprows when blocksize is smaller than sample size. Setting ``sample=blocksize``)Z delimiterrsample compressionZ include_pathFnamesZcommentrzSample is not large enough to include at least one row of data. Please increase the number of bytes in `sample` in the call to `read_csv`/`read_table`r0r4ZEOFzEOF encountered while reading header. Pass argument `sample_rows` and make sure the value of `sample` is large enough to accommodate that many rows of datazFiles already contain the column name: %s, so the path column cannot use this name. Please set `include_path_column` to a unique name.rcSsg|]}dd|DqS)cSsg|]}t|jqSr))r6daskvalues)r-Zdskr)r)r*r/}r0z*read_pandas...r))r-r>r)r)r*r/}r0zread_pandas..)r&r{r?rr)+rErUrzboolrNr;rsrrmaxrrr6 TypeErrordictr rAUTO_BLOCKSIZEr^rrr NotImplementedErrorencoderrcomputesplitdecoderostripr9r:rrQruZ ParserErrorr rrrpfloatr)$r$rrrrr sample_rowsr&assume_missingrinclude_path_columnr' reader_namekwr3Z lastskiprowZfirstrowr~rWZb_lineterminatorZb_outZb_samplerr?rr#Zneedrr=Z split_commentZnpartsZ head_kwargsr"rZr{r.r)r)r* read_pandass                    &     ra Read {file_type} files into a Dask.DataFrame This parallelizes the :func:`pandas.{reader}` function in the following ways: - It supports loading many files at once using globstrings: >>> df = dd.{reader}('myfiles.*.csv') # doctest: +SKIP - In some cases it can break up large files: >>> df = dd.{reader}('largefile.csv', blocksize=25e6) # 25MB chunks # doctest: +SKIP - It can read CSV files from external resources (e.g. S3, HDFS) by providing a URL: >>> df = dd.{reader}('s3://bucket/myfiles.*.csv') # doctest: +SKIP >>> df = dd.{reader}('hdfs:///myfiles.*.csv') # doctest: +SKIP >>> df = dd.{reader}('hdfs://namenode.example.com/myfiles.*.csv') # doctest: +SKIP Internally ``dd.{reader}`` uses :func:`pandas.{reader}` and supports many of the same keyword arguments with the same performance guarantees. See the docstring for :func:`pandas.{reader}` for more information on available keyword arguments. Parameters ---------- urlpath : string or list Absolute or relative filepath(s). Prefix with a protocol like ``s3://`` to read from alternative filesystems. To read from multiple files you can pass a globstring or a list of paths, with the caveat that they must all have the same protocol. blocksize : str, int or None, optional Number of bytes by which to cut up larger files. Default value is computed based on available physical memory and the number of cores, up to a maximum of 64MB. Can be a number like ``64000000`` or a string like ``"64MB"``. If ``None``, a single block is used for each file. sample : int, optional Number of bytes to use when determining dtypes assume_missing : bool, optional If True, all integer columns that aren't specified in ``dtype`` are assumed to contain missing values, and are converted to floats. Default is False. storage_options : dict, optional Extra options that make sense for a particular storage connection, e.g. host, port, username, password, etc. include_path_column : bool or str, optional Whether or not to include the path to each particular file. If True a new column is added to the dataframe called ``path``. If str, sets new column name. Default is False. **kwargs Extra keyword arguments to forward to :func:`pandas.{reader}`. Notes ----- Dask dataframe tries to infer the ``dtype`` of each column by reading a sample from the start of the file (or of the first file if it's a glob). Usually this works fine, but if the ``dtype`` is different later in the file (or in other files) this can cause issues. For example, if all the rows in the sample had integer dtypes, but later on there was a ``NaN``, then this would error at compute time. To fix this, you have a few options: - Provide explicit dtypes for the offending columns using the ``dtype`` keyword. This is the recommended solution. - Use the ``assume_missing`` keyword to assume that all columns inferred as integers contain missing values, and convert them to floats. - Increase the size of the sample using the ``sample`` keyword. It should also be noted that this function may fail if a {file_type} file includes quoted strings that contain the line terminator. To get around this you can specify ``blocksize=None`` to not split files into multiple partitions, at the cost of reduced parallelism. c s(d fdd }tj||d|_||_|S) NrrrrFc s&t|f||||||||| d | S)N) rrrrrr&rrr)r) rrrrrrr&rrrr'r$r)r*reads  zmake_reader..read)r$ file_type) rNrrrFFNF)READ_DOC_TEMPLATErsrHrE)r$rrrr)rr* make_readersrread_csvZCSV read_tableZ delimitedread_fwfz fixed-width) depend_oncKsD|"}|j|fi|Wdn1s,0Ytj|jSr)to_csvosr?normpath)rCZfilrr'fr)r)r* _write_csvs0rutf-8wtc  s<|r|durtd| dur"|} n| s2|r2tdtf||dd| pFi} ttdd|}|r t|fd|i| }t|jtj j j st d |d |fi}| d dd }t|fd|i| }dd <|ddD]}||fd|i}q|g}|g}ntt|f|||jd| }|d |d fig}| rPdd <|fddt|dd|ddD|r4| durt} |durt d|dt|dur| ddur| d|krtd|d| d|dur| ddur|| d<d dl}t|j|i| S|SdS)a Store Dask DataFrame to CSV files One filename per partition will be created. You can specify the filenames in a variety of ways. Use a globstring:: >>> df.to_csv('/path/to/data/export-*.csv') # doctest: +SKIP The * will be replaced by the increasing sequence 0, 1, 2, ... :: /path/to/data/export-0.csv /path/to/data/export-1.csv Use a globstring and a ``name_function=`` keyword argument. The name_function function should expect an integer and produce a string. Strings produced by name_function must preserve the order of their respective partition indices. >>> from datetime import date, timedelta >>> def name(i): ... return str(date(2015, 1, 1) + i * timedelta(days=1)) >>> name(0) '2015-01-01' >>> name(15) '2015-01-16' >>> df.to_csv('/path/to/data/export-*.csv', name_function=name) # doctest: +SKIP :: /path/to/data/export-2015-01-01.csv /path/to/data/export-2015-01-02.csv ... You can also provide an explicit list of paths:: >>> paths = ['/path/to/data/alice.csv', '/path/to/data/bob.csv', ...] # doctest: +SKIP >>> df.to_csv(paths) # doctest: +SKIP Parameters ---------- df : dask.DataFrame Data to save filename : string Path glob indicating the naming scheme for the output files single_file : bool, default False Whether to save everything into a single CSV file. Under the single file mode, each partition is appended at the end of the specified CSV file. Note that not all filesystems support the append mode and thus the single file mode, especially on cloud storage systems such as S3 or GCS. A warning will be issued when writing to a file that is not backed by a local filesystem. encoding : string, optional A string representing the encoding to use in the output file, defaults to 'ascii' on Python 2 and 'utf-8' on Python 3. mode : str Python write mode, default 'w' name_function : callable, default None Function accepting an integer (partition index) and producing a string to replace the asterisk in the given filename globstring. Should preserve the lexicographic order of partitions. Not supported when `single_file` is `True`. compression : string, optional a string representing the compression to use in the output file, allowed values are 'gzip', 'bz2', 'xz', only used when the first argument is a filename compute : bool If true, immediately executes. If False, returns a set of delayed objects, which can be computed at a later time. storage_options : dict Parameters passed on to the backend filesystem class. header_first_partition_only : boolean, default None If set to `True`, only write the header row in the first output file. By default, headers are written to all partitions under the multiple file mode (`single_file` is `False`) and written only once under the single file mode (`single_file` is `True`). It must not be `False` under the single file mode. compute_kwargs : dict, optional Options to be passed in to the compute method kwargs : dict, optional Additional parameters to pass to `pd.DataFrame.to_csv()` Returns ------- The names of the file written if they were computed right away If not, the delayed tasks associated to the writing of the files Raises ------ ValueError If `header_first_partition_only` is set to `False` or `name_function` is specified when `single_file` is `True`. Nz9name_function is not supported under the single file modezDheader_first_partition_only cannot be False in the single file mode.rc)rencodingnewlineF)Zpurerz8Appending data to a network storage system may not work.rwar#r|r)r name_functionZnumcs"g|]\}}||fiqSr)r))r-drr'Z to_csv_chunkr)r*r/r0zto_csv..zThe 'scheduler' keyword argument for `to_csv()` is deprecated andwill be removed in a future version. Please use the `compute_kwargs` argument instead. For example, df.to_csv(..., compute_kwargs={scheduler: z}) schedulerzJDiffering values for 'scheduler' have been passed in. scheduler argument: z via compute_kwargs: )rNrrrZ to_delayed open_filerzZfsfsspecZimplementationslocalZLocalFileSystemrreplacer Z npartitionsextendzip FutureWarningr;rr6r)rCfilenameZ single_filerrrrrrrZheader_first_partition_onlyZcompute_kwargsr'Z file_optionsZdfsZ first_filevalueZ append_modeZ append_filerrfilesrr)rr*rsq  (     r)_Frame)NNTFN)FNNNN) rNrrrFFNF) FrrNNTNNNN)Crcollections.abcriorwarningsrrrZhighlevelgraphrZlayersr r ImportErrorZfsspec.implementations.localrZnumpyrTZpandasrQZfsspec.compressionr Z fsspec.corer r rr Z fsspec.utilsrZpandas.api.typesrrrrrbaserbytesrcorerrZutilsrrrrrr<rMrrrrrrrrrrrrrrrrHr)r)r)r*s                  n 4a t EK"  >