o Nrf@sHddlmZddlZddlmZddlmZddlmZm Z m Z zddl Z Wn e y1dZ Ynwddl ZddlZddlmZddlmZddlmZdd lmZdd lmZdd lmZmZmZmZmZdd l m!Z!dd l"m#Z#ddl$m%Z%ddl&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1m2Z2Gddde+Z3     d@ddZ4ddZ5     dAddZ6dd Z7d!d"Z8d#d$Z9d%d&Z:e:Z; '  ( ) *    dBd+d,Ze'j?d0d1d2e>ej@d1d3Z@e>ejAd4d5ZAe>ejBd6d7ZBdd8d9d:ZC  ; <       dCd=d>ZDdd?lEmFZFeDjGeFjD_GdS)D) annotationsN)Mapping)BytesIO)catch_warnings simplefilterwarn)compr)get_fs_token_paths)open) open_files)infer_compression)CategoricalDtypeis_datetime64_any_dtypeis_float_dtypeis_integer_dtypeis_object_dtype)tokenize) read_bytes)flatten)dataframe_creation_dispatch)from_map)DataFrameIOFunction)clear_known_categories)delayed) asciitable parse_bytesc@s4eZdZdZddZeddZddZdd Zd S) CSVFunctionWrapperzg CSV Function-Wrapper Class Reads CSV data from disk to produce a partition (given a key). c Cs:||_||_||_||_||_||_||_||_| |_dSN) full_columns_columnscolnameheadheaderreaderdtypesenforcekwargs) selfrcolumnsr r!r"r#r$r%r&r)^/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/dask/dataframe/io/csv.py__init__-s  zCSVFunctionWrapper.__init__cCs*|jdur|jS|jr|j|jgS|jSr)rrr r'r)r)r*r(Cs zCSVFunctionWrapper.columnsc sfddjjDjkrSjr$jvr$jjg}nj}tjj|jjfddDjj S)zUReturn a new CSVFunctionWrapper object with a sub-column projection. csg|]}|vr|qSr)r).0c)r(r)r* Pz6CSVFunctionWrapper.project_columns..csi|]}|j|qSr))r$r-r,r)r* `r1z6CSVFunctionWrapper.project_columns..) r!r(r rrr"r#r%r&)r'r(r!r))r(r'r*project_columnsKs"  z"CSVFunctionWrapper.project_columnsc Cs|\}}}}|dur|j|tt|j|jjjf}nd}d}|j}|sD|dddur0d}| dd|dddurD| dd|sL| dd|j } d} |j durc|jr\d} n|j } | |d<t |j ||j||j| ||j| } | rz| |jS| S) NFnamesTskiprowsr"r skipfooterZusecols)r sortedlistr!cat categoriesr&copygetpoprrpandas_read_textr#r"r$r%r() r'partblockpathis_firstis_lastZ path_info write_headerZ rest_kwargsr(Zproject_after_readdfr)r)r*__call__esL       zCSVFunctionWrapper.__call__N) __name__ __module__ __qualname____doc__r+propertyr(r3rFr)r)r)r*r's  rTFc  Cst} |r||s| || || d|| fi|} |r*t| ||r>|r>t| jt|kr>td| j||r_|\} }} | |} | j di| 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 matchNr))r startswithrstripwriteseek coerce_dtypesr8r( ValueErrorindexassignpd Categorical from_codesnpfulllen)r#br"r&r$r(rDr%rAZbiorEr pathscoder)r)r*r>s"       r>c Csg}g}g}|jD]f}||vro|j|||kro|j|}||}t|r2t|r2||||fq t|r@t|r@||q z ||||||<Wq tyn}z||||f|||fWYd}~q d}~wwq |r|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* s z coerce_dtypes..cS t|dSNrstrxr)r)r* zcoerce_dtypes..)keyzAThe following columns also raised exceptions on conversion: %s zf Alternatively, provide `assume_missing=True` to interpret all unspecified integer columns as floats.cSr`rarbrdr)r)r*rfrg)ColumnFoundZExpectedz dtype={%s}z , css&|]\}}}|d|dVqdS)z: ''Nr))r.kv_r)r)r*r_s z{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_saThe 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 Exceptionjoinr7rformatfilterrQ)rEr$Z bad_dtypesZ bad_dateserrorsr/actualZdesiredr^exrqrsrprrZ dtype_msgrurvZdate_msgrulemsgr)r)r*rPsl           rPc  s|j} |jdgdj} ttr!fdd| D} | | } n| } | D]}d| |<q%t|j}tt |}tt |}tt |}|rw|\}dd|D}r[fdd|D}|j di|t jtjt|tdt|i}||f}t| rt|| d}g}|pd \}}tt|D]}||||r||nd ||||gqtt|d ||||| || ||d t|| ||||d d 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)includecs0g|]}t|tr|jdur|qSr) isinstancer<r r:)r.rm)specified_dtypesr)r*r0\sz)text_blocks_to_pandas..cSsg|]}|djqS))rA)r.rZr)r)r*r0sscsg|]}|qSr)r))r.p)path_converterr)r*r0us)dtype)rv)NNNzread-csvFT)metalabeltokenZenforce_metadataZproduces_tasksr))r$to_dictZ select_dtypesr(rr differencer8tupler block_maskblock_mask_lastrSrTrUrVrWZzerosrYintsetrrangerwrrr)r# block_listsr"r!r&r%rrA blocksizeurlpathr$Z categoricalsZknown_categoricalsZunknown_categoricalsrmr(blocksrBrCr r[partsir))rrr*text_blocks_to_pandas-sf &          , rccs8|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] Tcs|]}dVqdSFNr)r.ror)r)r*r_zblock_mask..rNr)rr@r)r)r*rsrccs8|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] csrrr)rr)r)r*r_rz"block_mask_last..NTr)rr)r)r*rsrcCs"d}t|||}t|tdS)N gA)rmin)Z total_memory cpu_countZ memory_factorrr)r)r*auto_blocksizesrcCsdd}tdur0ttdttj}t}Wdn1s"wY|r0|r0t||S|S)Niignore)psutilrrRuntimeWarningZvirtual_memorytotalrr)defaultZmemcpur)r)r*_infer_block_sizes    rrinferrc % Ksh|j} |durt|dkr|| d<nd}d| vr1|| d} d| d}|r0| t|d} n|} | r>t| tr>d} d| vsMd| vrQ| dd urQtd d D]}|| vrbt|d | qS| d drptd| t| dtr| d}}n'| ddurd}}nt | d}t |}t t t t|dt |}t| dt rtd| t| dtr| r| d| d}nd}|dkrt|d| dd}t|dkrt|dt|d}|dkrt}t|trt|}|r |r td|d}|tvrtd||r*|r*||kr*|dkr*td|}t|f| |||| d| p8i}| rH|\}}}| |f}n|\}}d}t|dtt fs[|g}|d urot|dro|dd}| dd}| d|durdnd}|durdnd}t|tr||7}| drg}|| D]7}|| d}t|dkrt|ddkr||dn||t||krnqn || t ||||}|sdn t|t|d  }|d ur|||krt||krtd!|durd"n||| }| } | !d#d| d$d%kr2d&| d$<z|t"|fd |i| }!Wnt#j$j%y\}"z d't|"vrWtd(|"d}"~"ww| rl| |!j&vrltd)| | d*i}#|#duryi}#|rt|#tr|!j&D]}$t'|!|$j(r|$|#vr|!|$)t*|!|$<qd+d,|D}t+||||!| ||#|||d- S).Nrlineterminatorr]encodingrirArRZ index_colFzKeywords 'index' and 'index_col' not supported, except for 'index_col=False'. Use dd.{reader_name}(...).set_index('my-index') instead)iterator 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)`r5rr"z)List of header rows not supported for dd. convertersrrb)modestorage_optionsz resolved to no filesrzWarning %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``) delimiterrsample compressionZ include_pathr4commentrzSample 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`r6ZengineZpyarrowr/ZEOFzEOF 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))r8daskvalues)r.Zdskr)r)r*r0z*read_pandas...r))r.r@r)r)r*r0rzread_pandas..)r%rrArr),rGrYencoderboolr<rQr{rrmaxrrr8 TypeErrordictr OSErrorr AUTO_BLOCKSIZErcrrrNotImplementedErrorrrcomputesplitdecoderwstripr;r=rrTr}Z ParserErrorr(rrrxfloatr)%r#rrrrr sample_rowsr%assume_missingrinclude_path_columnr& reader_nameZb_lineterminatorZ empty_blobkwZ lastskiprowZfirstrowr5rr[Zb_outZb_samplerrAr4r"Zneedrr?Z split_commentZnpartsZ head_kwargsr!r^rr/r)r)r* read_pandass4                   &    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. cs:       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_TEMPLATEr{rJrG)r#rrrr)rr* make_readersrpandasread_csv)backendnameZCSV read_tableZ delimitedread_fwfz fixed-width) depend_oncKsD|}|j|fi|Wdn1swYtj|jSr)to_csvosrAnormpath)rEZfilrr&fr)r)r* _write_csv!srutf-8wtc  s(|r |dur td| dur|} n| s|rtdtd||dd| p#i} ttdd|}|rt|fd|i| }|d |fi}d |vrM|n|d }|d dd d}t|fd|i| }dd <|ddD] }||fd|i}qo|g}|g}n9t|f|||jd| }|d |d fig}| rdd <| fddt |dd|ddD|r| durt} |durt d|dt |dur| ddur| d|krtd|d| d|dur| ddur|| d<d dl}t|j|i| S|S)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 You can also provide a directory name: >>> df.to_csv('/path/to/data') # doctest: +SKIP The files will be numbered 0, 1, 2, (and so on) suffixed with '.part': :: /path/to/data/0.part /path/to/data/1.part Parameters ---------- df : dask.DataFrame Data to save filename : string or list Absolute or relative filepath(s). Prefix with a protocol like ``s3://`` to save to remote filesystems. 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. encoding : string, default 'utf-8' A string representing the encoding to use in the output file. mode : str, default 'w' Python file mode. The default is 'w' (or 'wt'), for writing a new file or overwriting an existing file in text mode. 'a' (or 'at') will append to an existing file in text mode or create a new file if it does not already exist. See :py:func:`open`. 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, default True 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 : bool, 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 be True 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 :meth:`pandas.DataFrame.to_csv`. Returns ------- The names of the file written if they were computed right away. If not, the delayed tasks associated with writing the files. Raises ------ ValueError If ``header_first_partition_only`` is set to False or ``name_function`` is specified when ``single_file`` is True. See Also -------- fsspec.open_files Nz9name_function is not supported under the single file modezDheader_first_partition_only cannot be False in the single file mode.ri)rrnewlineF)Zpurerrawrer"rr)r name_functionnumcs"g|] \}}||fiqSr)r))r.drr&Z to_csv_chunkr)r*r0s"zto_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: r))rQrrrZ to_delayed open_filereplacer Z npartitionsextendzipr FutureWarningr<rr8r)rEfilenameZ single_filerrrrrrrZheader_first_partition_onlyZcompute_kwargsr&Z file_optionsdfsZ first_filevalueZ append_modeZ append_filerrfilesrr)rr*r's  ( r)_Frame)NNTFN)FNNNNr) FrrNNTNNNN)H __future__rrcollections.abcriorwarningsrrrr ImportErrornumpyrWrrTZfsspec.compressionrZ fsspec.corer r rr Z fsspec.utilsr Zpandas.api.typesr rrrrZ dask.baserZ dask.bytesrZ dask.corerZdask.dataframe.backendsrZdask.dataframe.io.iorZdask.dataframe.io.utilsrZdask.dataframe.utilsrZ dask.delayedrZ dask.utilsrrrr>rPrrrrrrrrrZregister_inplacerrrrrZdask.dataframe.corerrJr)r)r)r*s                  | 4a q WK"    L