B T\Hg @s4ddlmZmZmZddlmZddlmZmZm Z y ddl Z Wne k rXdZ YnXddl Z ddlZyddlmZdZWne k rdZYnXdd lmZmZdd lmZmZdd lmZmZmZmZdd lmZdd l m!Z!m"Z"ddl m#Z#m$Z$ddlm%Z%e$dkr,ddlm&Z&m'Z'm(Z(m)Z)nddl*m&Z&m'Z'm(Z(m)Z)d/ddZ+ddZ,d0ddZ-ddZ.e dk re,e de/e 0j1Z2e 3Z4e.e2e4Z5WdQRXndZ5e5ddddddddf dd Z6d!Z7d"d#Z8e8ej9d$d%Z9e8ej:d&d'Z:e8ej;d(d)Z;d*d+Zm?Z?e=j@e?j=_@dS)2)print_functiondivisionabsolute_import)BytesIO)warncatch_warnings simplefilterN)CategoricalDtypeTF) read_bytes open_files)seekable_filesfiles)PY2PY3Mappingunicode)delayed) asciitable parse_bytes)clear_known_categoriesPANDAS_VERSION) from_delayedz0.20.0)is_integer_dtypeis_float_dtypeis_object_dtypeis_datetime64_any_dtypec  Cst} |r"||s"| || || d|| f|} |rPt| ||rz|rzt| jt|krztd| j|n |r|| _|r|\} }} | |} | j f| t j tt| | | i} | S)aS Convert 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 all paths. See Also -------- dask.dataframe.csv.read_pandas_from_bytes rzColumns do not match)r startswithrstripwriteseek coerce_dtypeslistcolumns ValueErrorindexassignpd Categorical from_codesnpZfulllen)readerbheaderkwargsdtypesr% write_headerenforcepathZbiodfcolnamepathscoder:4lib/python3.7/site-packages/dask/dataframe/io/csv.pypandas_read_text&s$        r<c Csg}g}g}x|jD]}||kr|j|||kr|j|}||}t|rft|rf||||fqt|rt|r||qy||||||<Wqtk r}z"||||f|||fWdd}~XYqXqW|rp|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)z Coerce dataframe to dtypes safely Operates in place Parameters ---------- df: Pandas DataFrame dtypes: dict like {'x': float} N css|]\}}d||fVqdS)z - %s %rNr:).0cer:r:r; tsz coerce_dtypes..cSs t|dS)Nr)str)xr:r:r;uszcoerce_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|dS)Nr)rB)rCr:r:r;rDsZColumnZFoundZExpectedz dtype={%s}z , css |]\}}}d||fVqdS)z%r: '%s'Nr:)r>kv_r:r:r;rAsz{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;rAsaThe 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%r2rrappendrrastype Exceptionjoinsortedrformatfilterr&)r6r2Z bad_dtypesZ bad_dateserrorsr?ZactualZdesiredr@exrKrMrJrLZ dtype_msgrOrPZdate_msgZrulemsgr:r:r;r#RsZ    $     r#c s|j} |jdgdj} g} | } trjttrNfdd| D} | | } nttrjj dkrjg} | } x| D] } d| | <qpWt |j}t t dd}g}|pd\}}xt |D]\}}|sq|r||||f}nd}|||d ||| |d ||d }|||}|d dx4|d dD]$}||||||| |||dq WqW|r|r||jf|tjtt||i}t| rt|| d}t||S|SdS)a Convert blocks of bytes to a dask.dataframe or other high-level object 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. Can be ``None`` if ``collection==False`` kwargs : dict Keyword arguments to pass down to ``reader`` collection: boolean, optional (defaults to True) path : tuple, optional A tuple containing column name for path and list of all paths Returns ------- A dask.dataframe or list of delayed values category)Zincludecs0g|](}t|tr|jdk r|qS)N) isinstancegetr categories)r>rG)specified_dtypesr:r; sz)text_blocks_to_pandas..NT)pure)NNrF)r3r4r5skiprowsr)r4r5)rP)r2Zto_dictZ select_dtypesr%_HAS_CDTr\r differencer r^r$rr< enumeraterQcopypopr(r)r*r+r,Zzerosr-rr)r.Z block_listsr0headr1 collectionr4r_r5r2Z categoricalsZknown_categoricalsZunknown_categoricalsrGr%Zdelayed_pandas_read_textdfsr7r8iZblocksZ path_infor6Z rest_kwargsr/r:)r_r;text_blocks_to_pandassT                rlcCs"d}t|||}t|tdS)N gA)intmin)Z total_memory cpu_countZ memory_factor blocksizer:r:r;auto_blocksizesrrignoreiic  s|j} |dk r$t|dkr$|| d<nd}| r:t| tr:d} d| ksJd| krXtd| x$dD]} | | kr^td | | q^W| 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| dndt|ttfrvt|}|r|tkrtd|d}|tkr|tkrtd||r||kr|d krtd|}|}t|f||||| d| pi}| r8|\}}}r.fdd|D}| |f}n |\}}d}t|d tt fs^|g}| dd}| d|dkr~dnd}|dkrdnd}||||}|sd nt|t|d }|||krt||krtd|dkrdn |||}|t|f| }| r2| |jkr2td| | di}|dkrLi}|rt|trx<|jD]2}t||jrf||krf||t||<qfWt ||||| ||||d S) Nrlineterminatorr=r5r'Z index_colzGKeyword 'index' not supported dd.{0}(...).set_index('my-index') instead)iteratorZ chunksizez{0} not supported for dd.{1}ZnrowszThe 'nrows' keyword is not supported by `dd.{0}`. To achieve the same behavior, it's recommended to use `dd.{0}(...).head(n=nrows)`rbrr0z,List of header rows not supported for dd.{0}Z converterszWarning %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 delimiterrqsample compressionZ include_pathcsg|] }|qSr:r:)r>r5)path_converterr:r;r`^szread_pandas..namesZinferrzSample 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`zFiles already contain the column name: %s, so the path column cannot use this name. Please set `include_path_column` to a unique name.dtype)rir4r_r5)!__name__r-r\boolr&rVr]rnsetmaxroranger$ TypeErrordictrBrrr rcfilesNotImplementedErrorencoder tuplesplitrr%rr|rRfloatrl)r.urlpathrqrirtrwrvr4assume_missingstorage_optionsinclude_path_columnr1 reader_namekwrbZ lastskiprowZfirstrowZb_lineterminatorZb_outZb_samplevaluesr8r5ryr0ZneedpartsZnpartsrhr_r?r:)rxr; read_pandass            raQ 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. If ``None``, use a single block for each file. Can be a number like 64000000 or a string like "64MB" collection : boolean, optional Return a dask.dataframe if True or list of dask.delayed objects if False 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:tddddddddf fdd }tj||d|_||_|S)NTiFc s&t|f||||||||| d | S)N) rqrirtrwrvr4rrr)r) rrqrirtrwrvr4rrrr1)r.r:r;reads zmake_reader..read)r. file_type)AUTO_BLOCKSIZEREAD_DOC_TEMPLATErV__doc__r})r.rrrr:)r.r; make_readers  rread_csvZCSV read_tableZ delimitedread_fwfz fixed-widthc Ks"|}|j|f|WdQRXdS)N)to_csv)r6Zfilr1fr:r:r; _write_csvsrc strd} d} nd} d} d| } t|f|| | ||jd|p>i} ttdd|} | d | d fg}t| d kr|rdd <|fd d t | d d| d dD|rt|j |ddd | DS|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 ---------- filename : string Path glob indicating the naming scheme for the output files 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 compression : string or None String like 'gzip' or 'xz'. Must support efficient random access. Filenames with extensions corresponding to known compression algorithms (gz, bz2) will be compressed accordingly automatically sep : character, default ',' Field delimiter for the output file na_rep : string, default '' Missing data representation float_format : string, default None Format string for floating point numbers columns : sequence, optional Columns to write header : boolean or list of string, default True Write out column names. If a list of string is given it is assumed to be aliases for the column names header_first_partition_only : boolean, default False If set, only write the header row in the first output file index : boolean, default True Write row names (index) index_label : string or sequence, or False, default None Column label for index column(s) if desired. If None is given, and `header` and `index` are True, then the index names are used. A sequence should be given if the DataFrame uses MultiIndex. If False do not print fields for index names. Use index_label=False for easier importing in R nanRep : None deprecated, use na_rep mode : str Python write mode, default 'w' 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. 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 line_terminator : string, default '\n' The newline character or character sequence to use in the output file quoting : optional constant from csv module defaults to csv.QUOTE_MINIMAL quotechar : string (length 1), default '"' character used to quote fields doublequote : boolean, default True Control quoting of `quotechar` inside a field escapechar : string (length 1), default None character used to escape `sep` and `quotechar` when appropriate chunksize : int or None rows to write at a time tupleize_cols : boolean, default False write multi_index columns as a list of tuples (if True) or new (expanded format) if False) date_format : string, default None Format string for datetime objects decimal: string, default '.' Character recognized as decimal separator. E.g. use ',' for European data storage_options: dict Parameters passed on to the backend filesystem class. 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 Nwbzutf-8Zwtencoding)rwmoder name_functionZnumF)rarrr0csg|]\}}||fqSr:r:)r>dr)r1 to_csv_chunkr:r;r`szto_csv..) schedulercSsg|] }|jqSr:)r5)r>rr:r:r;r`s) rr]r Z npartitionsrrZ to_delayedr-extendzipcompute)r6filenamerrwrrrZheader_first_partition_onlyr1Zdefault_encodingrrrrjrr:)r1rr;rs*x     r)_Frame)NNTFN)TFNN)NNTNNF)AZ __future__rrriorwarningsrrrZpsutil ImportErrorZnumpyr,Zpandasr)Zpandas.api.typesr rcbytesr r Zbytes.compressionr rrZ compatibilityrrrrrZutilsrrrrrrrrrZpandas.types.commonr<r#rlrrRuntimeWarningZvirtual_memoryZtotalZ TOTAL_MEMrpZ CPU_COUNTrrrrrrrrrZcorerrr:r:r:r;sl         +U ]    B