U /ei @sddlZddlmZddlmZddlmZmZmZz ddl Z Wne k rXdZ YnXddl Z ddl ZddlmZddlmZddlmZddlmZdd lmZdd lmZmZmZmZmZdd lmZdd 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/m0Z0Gddde)Z1d?ddZ2ddZ3d@ddZ4ddZ5d d!Z6d"d#Z7d$d%Z8e8Z9dAd*d+Z:d,Z;d-d.Zd0d2Z>elCmDZDeBjEeDjB_EdS)CN)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) selfrcolumnsrr r!r"r#r$r%r(9/tmp/pip-unpacked-wheel-dbjnr7gq/dask/dataframe/io/csv.py__init__+s zCSVFunctionWrapper.__init__cCs*|jdkr|jS|jr$|j|jgS|jSr)rrrr&r(r(r)r'As  zCSVFunctionWrapper.columnsc sfddjjDjkr$SjrHjkrHjjg}n j}tjj|jjfddDjj S)zUReturn a new CSVFunctionWrapper object with a sub-column projection. csg|]}|kr|qSr(r(.0c)r'r(r) Nsz6CSVFunctionWrapper.project_columns..csi|]}|j|qSr()r#r,r+r(r) ^sz6CSVFunctionWrapper.project_columns..) r r'rrrr!r"r$r%)r&r'r r()r'r&r)project_columnsIs"  z"CSVFunctionWrapper.project_columnsc Cs|\}}}}|dk r6|j|tt|j|jjjf}nd}d}|j}|s|dddkr`d}| dd|dddk r| dd|s| dd|j } d} |j dk r|jrd} n|j } | |d<t |j ||j||j| ||j| } | r| |jS| S) NFnamesTskiprowsr!r skipfooterZusecols)rsortedlistr 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__csL        zCSVFunctionWrapper.__call__N) __name__ __module__ __qualname____doc__r*propertyr'r1rDr(r(r(r)r%s  rTFc  Cst} |r"||s"| || || d|| f|} |rPt| ||rx|rxt| jt|krxtd| j||r|\} }} | |} | j f| 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_codesnpfulllen)r"br!r%r#r'rBr$r?ZbiorCrpathscoder(r(r)r<s"        r<c Csg}g}g}|jD]}||kr|j|||kr|j|}||}t|rdt|rd||||fqt|rt|r||qz||||||<Wqtk r}z"||||f|||fW5d}~XYqXq|rl|rd ddt |dddD} d| } d } nd } d } t |d dd}t d d dg|} dd 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)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.cSs t|dSr^r_rar(r(r)rcrdZColumnFoundZExpectedz 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]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#rrappendrr astype Exceptionjoinr5rformatfilterrO)rCr#Z bad_dtypesZ bad_dateserrorsr.actualZdesiredr\exrmrorlrnZ dtype_msgrqrrZdate_msgZrulemsgr(r(r)rNsp    "      rNc  s|j} |jdgdj} ttrBfdd| D} | | } n| } | D] }d| |<qJt|j}tt |}tt |}tt |}|r|\}dd|D}rfdd|D}|j f|t jtjt|tdt|i}||f}t| rt|| d}g}|p d \}}tt|D]2}||||r8||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|jdk r|qSr) isinstancer:r r8)r-ri)specified_dtypesr(r)r/[sz)text_blocks_to_pandas..cSsg|]}|djqS))r?)r-rXr(r(r)r/rscsg|] }|qSr(r()r-p)path_converterr(r)r/ts)dtype)rr)NNNzread-csvFT)metalabeltokenZenforce_metadataZproduces_tasks)r#Zto_dictZ select_dtypesr'rr differencer6tupler block_maskblock_mask_lastrQrRrSrTrUzerosrWintsetrrangersrrr)r" block_listsr!r r%r$rr? blocksizeurlpathr#Z categoricalsZknown_categoricalsZunknown_categoricalsrir'blocksr@rArrYpartsir()rrr)text_blocks_to_pandas,sj&          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-rkr(r(r)r]szblock_mask..rNr(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]sz"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_blocksizesrc CsPd}tdk rLt"tdttj}t}W5QRX|rL|rLt||S|S)Niignore)psutilrrRuntimeWarningZvirtual_memorytotalrr)defaultZmemcpur(r(r)_infer_block_sizes   rrinferrc $ Ks(|j} |dk r$t|dkr$|| d<nd}| r:t| tr:d} d| ksJd| krZtd| dd D]} | | kr^t| d | q^| d drtd | t| d tr| d }}}nR| d dkrd}}}n6t| d }t |}t tt t|dt|}t| dt r.t d| t| dtrZ| rZ| d| d}nd}|dkrt|d| dd}t|dkrt|dt|d}|dkrt}t|trt|}|r|rtd|d}|tkrtd||r"|r"||kr"|dkr"td|}|}t|f||||| d| pFi}| rf|\}}}| |f}n |\}}d}t|dtt fs|g}|dkrt|dr|dd}| dd}| d|dkrdnd}|dkrdnd}| drzg}||D]p}|| d}t|dkrVt|ddkr`||dn ||t||krqqn||||}|sdnt|t|d }|dk r|||krt||krtd t|tr||7}|dkrd!n |||}| }|!d"dz|t"|fd |i|} WnBt#j$j%k rx}!zd#t|!krftd$|!W5d}!~!XYnX| r| | j&krtd%| | d&i}"|"dkri}"|rt|"tr| j&D]2}#t'| |#j(r|#|"kr| |#)t*| |#<qd'd(|D}t+|||| | ||"|||d) S)*Nrlineterminatorr[r?rPZ index_colz7Keywords 'index' and 'index_col' not supported. Use dd.z#(...).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)`r3rr!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_pathFr2commentrzSample 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`rdr4EOFzEOF 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/sz*read_pandas...r()r-r>r(r(r)r/szread_pandas..)r$rr?rr),rErWrboolrOr:rwrrmaxrrr6 TypeErrordictrOSErrorr AUTO_BLOCKSIZEr`rrrNotImplementedErrorencoderrcomputesplitdecodersstripr9r;rrRryZ ParserErrorr'rrrtfloatr)$r"rrrrr sample_rowsr$assume_missingrinclude_path_columnr% reader_namekwr3Z lastskiprowZfirstrowrrYZb_lineterminatorZb_outZb_samplerr?r2r!Zneedrr=Z split_commentZnpartsZ head_kwargsr r\rr.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_TEMPLATErwrHrE)r"rrrr(rr) make_readersrpandasread_csv)backendnameZCSV read_tableZ delimitedread_fwfz fixed-width) depend_onc Ks,|}|j|f|W5QRXtj|jSr)to_csvosr?normpath)rCZfilrr%fr(r(r) _write_csvsrutf-8wtc  s|r|dk rtd| dkr"|} n| s2|r2tdtf||dd| pFi} ttdd|}|rt|fd|i| }|d |f}|d dd }t|fd|i| }dd <|d dD]}||fd|i}q|g}|g}npt|f|||jd| }|d |d fg}| r,dd <| fddt |d d|d dD|r | dkrrt} |dk rt d|dt |dk r| ddk r| d|krtd|d| d|dk r| ddkr|| d<d dl}t|j|| 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 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.rf)rencodingnewlineF)Zpurerrwar!rr)r name_functionnumcsg|]\}}||fqSr(r()r-drr%Z to_csv_chunkr(r)r/szto_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: )rOrrrZ to_delayed open_filereplacer Z npartitionsextendzipr FutureWarningr:rr6r)rCfilenameZ single_filerrrrrrrZheader_first_partition_onlyZcompute_kwargsr%Z file_optionsdfsZ first_filevalueZ append_modeZ append_filerrfilesrr(rr)rs  (      r)_Frame)NNTFN)FNNNN) rNrrrFFNF) FrrNNTNNNN)Frcollections.abcriorwarningsrrrr ImportErrorZnumpyrUrrRZfsspec.compressionrZ fsspec.corerr rr Z fsspec.utilsr Zpandas.api.typesr r rrrZ dask.baserZ dask.bytesrZ dask.corerZdask.dataframe.backendsrZdask.dataframe.io.iorZdask.dataframe.io.utilsrZdask.dataframe.utilsrZ dask.delayedrZ dask.utilsrrrr<rNrrrrrrrrrZregister_inplacerrrrrZdask.dataframe.corerrHr(r(r(r)s                 } 4a q IK"   K