>ieddlmZddlZddlmZddlmZddlmZm Z m Z  ddl Z n #e $rdZ YnwxYwddl 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/ddl0m1Z1m2Z2Gdde+Z3 d5dZ4dZ5 d6dZ6dZ7dZ8dZ9dZ:e:Z; d7d$Ze'j?d'd()e>ej@d(d*Z@e>ejAd+d,ZAe>ejBd-d.ZBdd/d0ZC d8d3ZDdd4lEmFZFeDjGeFjD_GdS)9) 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:eZdZdZdZedZdZdZdS)CSVFunctionWrapperzg CSV Function-Wrapper Class Reads CSV data from disk to produce a partition (given a key). c ||_||_||_||_||_||_||_||_| |_dSN) full_columns_columnscolnameheadheaderreaderdtypesenforcekwargs) selfr!columnsr#r$r%r&r'r(r)s 5lib/python3.11/site-packages/dask/dataframe/io/csv.py__init__zCSVFunctionWrapper.__init__-sH)        cZ|j|jS|jr|j|jgzS|jSr )r"r!r#)r*s r,r+zCSVFunctionWrapper.columnsCs6 = $ $ < 2=DL>1 1}r.c :fdjjDjkrSjr jvrjjgz}n j}tjj|jjfdDjj S)zUReturn a new CSVFunctionWrapper object with a sub-column projection. cg|]}|v| Sr2).0cr+s r, z6CSVFunctionWrapper.project_columns..Ps@@@1<<1<<z6CSVFunctionWrapper.project_columns..`s! 0 0 01Q A 0 0 0r.) r$r+r#rr!r%r&r(r))r*r+r$s`` r,project_columnsz"CSVFunctionWrapper.project_columnsKs A@@@di/@@@ dl " "K < &DL779W ~56DD9W%D!    L  K K 0 0 0 0 0 0 0 L K   r.c x|\}}}}|?|j|tt|j|jjjf}nd}d}|j}|sZ|ddd}| dd|dd| dd|s| dd|j } d} |j |jrd} n |j } | |d<t|j ||j||j| ||j| } | r | |jS| S) NFnamesTskiprowsr%r skipfooterusecols)r#sortedlistr$cat categoriesr)copygetpopr!r"pandas_read_textr&r%r'r(r+) r*partblockpathis_firstis_last path_info write_header rest_kwargsr+project_after_readdfs r,__call__zCSVFunctionWrapper.__call__ess)-&tXw   tDIdl37BCCDDII I k&&((  0w--5# OOJ - - -x++7$/// 0 OOL$ / / /#" = ${ 1&*""-)0 I& K  K  K   L      $dl# # r.N) __name__ __module__ __qualname____doc__r-propertyr+r8rPr2r.r,rr'sf ,X   466666r.rTFc zt} |r<||s| || || d|| fi|} |rt | ||r=|r;t | jt |krtd| j||rh|\} }} | |} | j di| tj tjt!| | | 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 matchr2)r startswithrstripwriteseek coerce_dtypesr?r+ ValueErrorindexassignpd Categorical from_codesnpfulllen)r&br%r)r'r+rLr(rHbiorOr#pathscodes r,rErEsE@ ))CALL99 &IIaLLLHHQKKK   v  B "b&!!!F7FRZ 0 0DMM A A/WEEE  #u{{4   RY  11"'#b''42H2H%PPQ   Ir.cg}g}g}|jD] }||vr|j|||kr|j|}||}t|r(t|r||||fkt |r%t |r|| ||||||<#t$r:}||||f|||fYd}~d}~wwxYw|r|r7d dt|dD} d| z} d} nd} d} t|d }tgd |} d d d |Dz} d | | | | }nd}|r=|rdnd}d d|D}d ||}nd}|s|rz coerce_dtypes..sH&*a!Q!!A!!r.c,t|dSNrstrxs r,zcoerce_dtypes..s3qQRt99r.)keyzAThe following columns also raised exceptions on conversion: %s zf Alternatively, provide `assume_missing=True` to interpret all unspecified integer columns as floats.c,t|dSrorprrs r,rtzcoerce_dtypes..sc!A$iir.)ColumnFoundExpectedz dtype={%s}z , c3.K|]\}}}|d|dVdS)z: ''Nr2)r3kv_s r,rmz coerce_dtypes..sE4 4 )AqqOOqOOO4 4 4 4 4 4 r.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  c3 K|] }d|zV dS)z- %sNr2)r3r4s r,rmz coerce_dtypes..s&77!777777r.aThe 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 Exceptionjoinr>rformatfilterr\)rOr' bad_dtypes bad_dateserrorsr4actualdesiredrlexrrrr dtype_msgrrdate_msgrulemsgs r,r[r[sJI F Z** ;;29Q<6!944Yq\FQiGf%% **:7*C*C *!!1fg"67777 (( *-DW-M-M *   ####*qELL33BqEE ***%%q&'&:;;;MM1a&))))))))*#  .4VATAT.U.U.UB(JEEJ=  J,?,?@@@ :::JGG,"3"34 4 -74 4 4 # #     &uhe& T T    %.xx3yy77Y77777 & &d& & & Yx(O IIfTIx#899 : : oo s.$C D/DDc L|j} |dgj} t t r$fd| D} | | } n| } | D]}d| |<t|j}tt|}tt|}tt|}|r|\}d|D}rfd|D}|j di|tjt!jt%|t&t)|i}||f}t%| rt+|| }g}|pd\}}t-t%|D]7}||||r||nd ||||g8t1t3|d ||||| || ||d t5|| ||||d d S)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)includecg|]F}t|tr|jD|GSr ) isinstancerCrrA)r3r}specified_dtypess r,r5z)text_blocks_to_pandas..\s`   *..q113CDD !$$Q''2> ?>>r.c(g|]}|djS))rH)r3res r,r5z)text_blocks_to_pandas..ss+++q1+++r.c&g|] }|Sr2r2)r3ppath_converters r,r5z)text_blocks_to_pandas..us#6661^^A&&666r.)dtype)r)NNNzread-csvFT)metalabeltokenenforce_metadataproduces_tasksr2)r'to_dict select_dtypesr+rr differencer?tupler block_maskblock_mask_lastr^r_r`rarbzerosrdintsetrrangerrrr)r& block_listsr%r$r)r(rrH blocksizeurlpathr' categoricalsknown_categoricalsunknown_categoricalsr}r+blocksrIrJr#rgpartsirs ` @r,text_blocks_to_pandasr-sL[ " "F %%zl%;;CL"G,, ,    !    ,667IJJ+"q 4<  G 7;'' ( (FZ ,,--HOK0011G  "&++F+++  76666666Et{  22HSYYc222CJJ     G%d1EFFF E)\NGU 3v;;  XX fQiU!<qhqk7ST:VWWWW             vw$ JJ#   r.c#VK|D]#}|sdVd|ddDEd{V$dS)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] Tc3K|]}dVdSFNr2r3rs r,rmzblock_mask..s"--aE------r.rNr2rrGs r,rrsf..   --59----------- ..r.c#VK|D]#}|sd|ddDEd{VdV$dS)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] c3K|]}dVdSrr2rs r,rmz"block_mask_last..s"..aE......r.NTr2rs r,rrse  ..5":.......... r.cjd}t||z|z }t|tdS)N gA)rmin) total_memory cpu_count memory_factorrs r,auto_blocksizers4MLI- =>>I y#d)) $ $$r.c d}tzt5tdttjj}tj}dddn #1swxYwY|r|rt||S|S)Niignore)psutilrrRuntimeWarningvirtual_memorytotalrr)defaultmemcpus r,_infer_block_sizersG     % % > 2 2 2'))/C"$$C % % % % % % % % % % % % % % %  ,3 ,!#s++ + NsAA%%A),A)rinferrc  |j} |t|dkr|| d<nd}| rt| trd} d| vsd| vr&| ddurt dd D]} | | vrt | d | | d dr"t d | t| d tr| d x}}n| d dx}}nzt| d }t|}tttt|dzt|z }t| dtrtd| t| dtr,| r*| d| d}nd}|dkrRt|d| d}t|dkrt!|dt#|d}|dkrt$}t|t&rt)|}|r|rt+d|zd}|t,vrt/d|z|r|r||kr|dkrt+d|}|}t3|f||||| d| pi}| r |\}}}| |f}n|\}}d}t|dt4tfs|g}|dur5t|dr |dd}| dd}| d|dnd}|dnd}t|tr||z }| drg}||D]}|| d}t|dkrYt|ddkr?||dn||t||krnn*||t||z||z}|sdn%t|t|d z }|dur+|||zkr"t||krt d|d n |||z}| }|!d!d|d"d#krd$|d"< |tE|fd |i|} n>#tFj$j%$r'}!d%t'|!vrt d&|!d}!~!wwxYw| r| | j&vrt d'| z| d(i}"|"i}"|r`t|"trK| j&D]C}#tO| |#j(r'|#|"vr#| |#)tT| |#<Dd)|D}tW|||| | ||"|||* S)+NrlineterminatorrjrHr] 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.nrowszThe 'nrows' keyword is not supported by `dd.{0}`. To achieve the same behavior, it's recommended to use `dd.{0}(...).head(n=nrows)`r;rr%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 compression include_pathr:commentrzSample 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`r.r<enginepyarrowr4EOFzEOF 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.rc&g|]}d|DS)cZg|](}t|j)Sr2)r?daskvalues)r3dsks r,r5z*read_pandas...s,8883tCHOO%%&&888r.r2)r3rGs r,r5zread_pandas..s' M M MU88%888 M M Mr.)r(rrHrr),rQrdrboolrCr\rrrmaxrrr? TypeErrordictr OSErrorr AUTO_BLOCKSIZErqrrr NotImplementedErrorencoderrcomputesplitdecoderstriprBrDrr_r ParserErrorr+rrrfloatr)$r&rrrrr sample_rowsr(assume_missingrinclude_path_columnr) reader_namekw lastskiprowfirstrowr;rrgb_lineterminatorb_outb_samplerrHr:r%needrrF split_commentnparts head_kwargsr$rlrr4s$ r, read_pandasr s/K!c.&9&9Q&>&>#1  %z*=tDD%$&v&**["9"9"F"F Y   (II <<GG+GGHH H  zz'4    $VK00    &**Z((#.. F!'J!7!77 hh J   '!"" hh vzz*--..(mm s5X!23344s8}}DEE&**X&&--SQKQQRRR&**\**D116IL11556I4PPg"7WWW   u::??W;;;<< <(a11 I" )S!!+ ** [  ),7 7    %!"G+"UVVVV F 2 2{a7G7G  +   %,,..  "(     b  E"'&%#^4 & fQi% / /5S^^!9Q<'')) JJw % %E ZZU]'' E EF11AD&#F zz) NN#344  D KKMM// 90E0EFFM=!!A%%}Q'((1,,LLq!1!7!7!9!9!@!@!B!BCCC T"""5zzD  ! c+"4hoFF    @QQUceBi-.@.@!@F Uv d(:::s8}}PV?V?V A   NSSh:J(JF++--KOOL$'''x  I-- # H vgh''JJ{JkJJ 9  CFF??H    3t| C C 68K L   zz'2..0*%5t<<0 0 0AQ .. 01>> 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. cv dfd }t|||_||_|S) NrrrrFc 4t |f||||||||| d | S)N) rrrrrr(rrr)r ) rrrrrrr(rrrr)r&s r,readzmake_reader..readsN     )##)+ 3      r.)r& file_type rNrrrFFNF)READ_DOC_TEMPLATErrTrQ)r&rrrs` r, make_readerrsf!      8%++;)+TTDLDM Kr.pandasread_csv)backendnameCSV read_table delimitedread_fwfz fixed-width) depend_onc |5}|j|fi|dddn #1swxYwYtj|jSr )to_csvosrHnormpath)rOfilrr)fs r, _write_csvr#s  !v 7  CH % %%s ""utf-8wtc R |r|td| |} n| s|rtdtd||dd| pi} ttd|}|rt |fd|i| }|d |fi }d |vr|n|d z}|d dd d}t |fd|i| }d d <|ddD]}||fd|i }|g}|g}ntt|f|||jd| }|d |d fi g}| rd d <| fdt|dd|ddD|r| t} |td|dt|V| dA| d|kr(td|d| d|| d|| 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.rv)rencodingnewlineF)purerrawrsr%rr)r name_functionnumc*g|]\}}||fiSr2r2)r3dr"r) to_csv_chunks r,r5zto_csv..s1 N N Nda\\!Q ) )& ) ) N N Nr.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: r2)r\rrr# to_delayed open_filereplacer npartitionsextendzipr FutureWarningrCrr?r)rOfilename single_filer'rr,rrr1rheader_first_partition_onlycompute_kwargsr) file_optionsdfs first_filevalue append_mode append_filer/rfilesrr0s ` @r,rrsd@V}0TUUU"*&1## ( [  R     b L :E222L --//C xCCdClCC  SVZ::6::!Tkkddtcz !))#r22::3CC KK{KlKK  xQRR L LA LKKK5KFKKEE   '      ,s1vuQx::6::; & %$F8  N N N N Nc#abb'596M6M N N N     !!VVN  ZLUZZZ      !"";//;"";//9<<I'0II'5'9'9+'F'FII   ^%7%7 %D%D%L*3N; ' LDL&;N;;<<< r.)_Frame)NNTFN)FNNNNr) Fr$r%NNTNNNN)H __future__rrcollections.abcriorwarningsrrrr ImportErrornumpyrbrr_fsspec.compressionr fsspec.corer r r3r fsspec.utilsr pandas.api.typesrrrrr dask.baser dask.bytesr dask.corerdask.dataframe.backendsrdask.dataframe.io.iordask.dataframe.io.utilsrdask.dataframe.utilsr dask.delayedr dask.utilsrrrrEr[rrrrrrr rrregister_inplacerrrr#rdask.dataframe.corerDrTr2r.r,rZs"""""" ######7777777777MMMM FFF$$$$$$******))))))""""""******!!!!!!??????))))))777777777777 ........ttttt,tttx    1111hXXXB   nnnnb . . .   %%%   #"$$   KKKK\HVD  7 & 7     +bk:u-- / / [ k B B ;r{J > >&*&&&&&    $HHHHV'&&&&& s '11