biZddlmZddlZddlmZddlmZddlmZddl m Z m Z m Z ddl mZ ddlZddlZddlZddlmZdd 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.ddl/m0Z0ddl1m2Z2m3Z3 d*dZ4dZ5 d+dZ6dZ7dZ8dZ9dZ:dZ;e;Z< d,dZ=dZ>dZ?e,jdd !e?ejd d"ZAe?ejd#d$ZBe?ejd%d&ZCdd'd(ZD d-d)ZEy#e$rdZYwxYw).) annotationsN)Mapping)partial)BytesIO)catch_warnings simplefilterwarn)convert_legacy_task)compr)get_fs_token_paths)open) open_files)infer_compression)CategoricalDtypeis_datetime64_any_dtypeis_float_dtypeis_integer_dtypeis_object_dtype) read_bytes)flatten)dataframe_creation_dispatch)clear_known_categories)delayed) asciitable parse_bytesc 8t} |r0|j|js| j|| j|| j d|| fi|} |r t | ||r:|r8t | jt |k7rtd| j||rf|\} }} | j|} | jdi| tjjtjt!| | | 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_dtypeslistcolumns ValueErrorindexassignpd Categorical from_codesnpfulllen)readerbheaderkwargsdtypesr$ write_headerenforcepathbiodfcolnamepathscodes P/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/dask/dataframe/io/csv.pypandas_read_textr<(s@ )CALL9 &IIaLHHQK  v B b&!7RZZ 0DM A/WEE #u{{4  RYY 11"''#b'42H%PQ  Icg}g}g}|jD]}||vs|j|||k7s|j|}||}t|r t|r|j |||f]t |rt |r|j | ||j||||<|r|r,djdt|dD} d| z} d} nd} d} t|d }tgd |} d d jd |Dz} dj| | | | }nd}|r2|rdnd}djd|D}dj||}nd}|s|r0ddz}d|jtd||gz}t|y#t$r2}|j |||f|j ||fYd}~d}~wwxYw)zCoerce dataframe to dtypes safely Operates in place Parameters ---------- df: Pandas DataFrame dtypes: dict like {'x': float} N c32K|]\}}d|d|yw)z- z Nr).0ces r; z coerce_dtypes..~s&&*a"QCtA5!sct|dSNrstrxs r;zcoerce_dtypes..s3qQRt9r=)keyzAThe following columns also raised exceptions on conversion: %s zf Alternatively, provide `assume_missing=True` to interpret all unspecified integer columns as floats.ct|dSrFrGrIs r;rKzcoerce_dtypes..sc!A$ir=)ColumnFoundExpectedz dtype={%s}z , c34K|]\}}}|d|dyw)z: ''Nr)rAkv_s r;rDz coerce_dtypes..s'4 )Aqqe3qcO4 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  c3&K|] }d|z yw)z- %sNr)rArBs r;rDz coerce_dtypes..s7!7saThe 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%)r7r2 bad_dtypes bad_dateserrorsrBactualdesiredrCexrXrZrWrY dtype_msgr]r^date_msgrulemsgs r;r"r"\s-JI F ZZ* ;299Q<6!94YYq\FQiGf%*:7*C!!1fg"67 (-DW-M   #*qELL3BqE*& .4VAT.UB(JEJ=  J,?@ :JG,"3"34 -74 #     &uhe& T  %x3yy7Y77 & &d& & Yx(O IIfTIx#89 : o o!*%%q&'&:;MM1a&))*sF## G,'GGc |jj} |jdgj} t |t r[| D cgc]>} t |j | tr|j | j| @} } | j| }n| }|D]} d| | < t|j}tt|}tt|}tt|}|r|\}}|Dcgc]}|jdj }}|r|Dcgc] }|| }}|j"d i|t$j&j)t+j,t/|t0t3|i}||f}t/|r t5||}g}|xsd\}}t7t/|D]#}|j9|r||nd||||g%t;j<t?t@|||||| ||| |||d d Scc} wcc}wcc}w) 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)includeNr)dtype)r^)NN) full_columnsr8headr0r.r2r4r1 blocksizezread-csvF)metalabelenforce_metadatar)!r2to_dict select_dtypesr$ isinstancergetr categories differencer#tupler block_maskblock_mask_lastargsr5r'r(r)r*r+zerosr-intsetrranger_ddfrom_mapr _read_csv)r. block_listsr0rur1r4specified_dtypesr5rvurlpathr2 categoricalsrTknown_categoricalsunknown_categoricalsr$blocksis_firstis_lastr8path_converterr/r9ppartsis r;text_blocks_to_pandasrsWL[[ "F %%zl%;CCL"G," *..q13CD $$Q'22>   ,667IJ+"q 4<< G 7;' (FZ ,-HOK01G "&)/0A00 0561^A&6E6t{{ 22HHSYc2CJ   %d1EF E)\NGU 3v; M %eAhT8A; KLM ;;     # [ .16s AH6 H;5Ic |\} } }| &|| t||jjf}nd}d}| j}| sJ|j ddd}|j dd|j dd|j dd|s|j dd|}d}|||k7r | rd}n|}||d<t |||||||| | }|r||S|S) NFnamesTskiprowsr0r skipfooterusecols)rccatr~copyr}popr<)blockpartr$r.r0r2rur8rtr4r1rvr5rr path_infor3 rest_kwargs_columnsproject_after_readr7s r;rr(s #D(G    4=$$// 0  L++-K  ??7D ) 1L D) ??8Q ' 3 OOHd +  d+Hw,6 "& H%-K "   B'{ Ir=c#TK|D]}|sdd|ddDEd{y7w)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] Tc3 K|]}dywFNrrArVs r;rDzblock_mask..{s-aE- Nrrrs r;rros;.  -59--- . .s (&(c#TK|D]}|sd|ddDEd{dy7 w)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] c3 K|]}dywrrrs r;rDz"block_mask_last..s.aE.rNTrrs r;rr~s; .5":...  /s (& (cRd}t||z|z }t|tdS)N gA)rmin) total_memory cpu_count memory_factorrvs r;auto_blocksizers,MLI- =>I y#d) $$r=cd}tet5tdttjj }tj }dddrr t||S|S#1swYxYw)Niignore)psutilrrRuntimeWarningvirtual_memorytotalrr)defaultmemcpus r;_infer_block_sizersmG    % > 2'')//C""$C % 3!#s+ + N % %s AA00A9c   |j} |t|dk(r|| d<nd}d| vr9|j| d} dj| d}|r| t|d} n|j} | rt| trd} d| vsd| vr| j dd ur t d d D]}|| vst |d | | j d rt dj| t| j dtr| j dx}}np| j ddx}}nZt| j d}t|}tttt|dzt|z }t| j dtrtd| t| j dtr$| r"| j dj | d}nd}|dk(r;t!|d| d}t|dk(rt#|dt%|d}|dk(rt&}t|t(r t+|}|r|rt-d|zd}|t.vrt1d|z|r|r||kr|dk7r t-d|}t3|f| |||| d| xsi}| r |\}}}| |f}n|\}}d}t|dt4tfs|g}|d ur$t|dr|ddj7}| j d}| j d|dnd}|| d<|dnd}t|tr||z }| j drg}|j9| D]}|j;j9| j d}t|dkDrBt|ddkDrB|j=|dj?jn|j=|t||kDsn$n"|j9| t||z||z}|sdnt|t|d  z }|d ur!|||zkrt||k\r t d!|d"n||| z}| jA} | jCd#d| j d$d%k(rd&| d$< |tE|fd |i| }!| r| |!jLvrt d)| z| j d*i}#|#i}#|rZt|#trJ|!jLD];}$tO|!|$jPs|$|#vs!|!|$jStT|!|$<=|D%&'(cgc]L}%|%D&'(cgc]:}&|&jVjYD'(cgc]\}'}(t[|'|(ic}(}'<c}(}'}&N}}'}&}%}(t]||||!| ||#|||+ S#tFjHjJ$r}"d't)|"vr t d(|"d}"~"wwxYwcc}(}'wcc}(}'}&wcc}(}'}&}%w),Nrlineterminatorr?encodingrMr5r& 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)`rrr0z)List of header rows not supported for dd. convertersinferrb)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``) delimiterrvsample compression include_pathrcommentrzSample 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=renginepyarrowrBEOFzEOF 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.rs)r4rr5rvr)/__name__r-encoder|boolr}r%rdrrmaxrrr# TypeErrordictr OSErrorrAUTO_BLOCKSIZErHrr r NotImplementedErrorrrcomputesplitdecoder_striprrrr(rh ParserErrorr$rrsr`floatdaskitemsr r))r.rrvrrr sample_rowsr4assume_missingrinclude_path_columnr1 reader_nameb_lineterminator empty_blobkw lastskiprowfirstrowrrr9b_outb_samplevaluesr5rr0needrr split_commentnparts head_kwargsrurCrrBrdskrTtss) r; read_pandasrs//K!c.&9Q&>#1 V)00 1CDYYvj12  0J0AB )002z*=tD$&v&**["9"F Y  (I <t#9+GH HIzz' $VK0   &**Z(#.!'J!77 h J  '!"" h vzz*-.(m s5X!234s8}DE&**X&-CK=QRR&**\*D16IL1556I4Pg"7W   u:?WI%:;< <(a1 I" )S! * [  ),7 7  %!"G+"UVVV F 2{a7G  +   "(    b E"'&%#^4 & fQi% /5S^!9Q<'') JJw E ZZU]' EFF81AD&#F zz)NN#34 D KKM// 90EFM=!A%}Q'(1,LLq!1!7!7!9!@!@!BC T"5zD   c+"4hoF  QUceBi-.@!@F Uv d(::s8}PV?V A  NSh:J(JF++-KOOL$'x I- # H gh'J{JkJ 3t|| C 68K L  zz'2.*%5t< 0AQ .1>> 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. cl dfd }tj|||_||_|S)Nc 2t |f||||||||| d | S)N) rvrrrrr4rrr)r) rrvrrrrr4rrrr1r.s r;readzmake_reader..readsD     )##)+ 3    r=)r. file_type rNrirFFNF)READ_DOC_TEMPLATErd__doc__r)r.rrrs` r; make_readerrsJ! 8%++;)+TDLDM Kr=pandasread_csv)backendnameCSV read_table delimitedread_fwfz fixed-width) depend_onc |5}|j|fi|dddtjj|jS#1swY2xYw)N)to_csvosr5normpath)r7filr r1fs r; _write_csvrsH  !v 77  CHH %%s AAc |j}|r | td| |} n| s |r tdtd||dd| xsi} ttd}|j }|rxt |fd|i| }||d |fi| }d |vr|n|d z}|jd djd d}t |fd|i| }d| d <|ddD]}|||fd|i| }|g}|g}nnt|f|||jd| }||d |d fi| g}| rd| d <|jt|dd|ddDcgc]\}}|||fi| c}}|r| t} |td|dt|E| jd4| jd|k7r td|d| jd|| jd|| d<d dl}t!|j"|i| S|Scc}}w)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.rM)rrnewlineF)purerrawrJr0rr )r name_functionnumzThe '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)optimizer%rrr to_delayed open_filereplacer npartitionsextendzipr FutureWarningr}rr#r)r7filename single_filerrrrrrrheader_first_partition_onlycompute_kwargsr1 file_options to_csv_chunkdfs first_filevalue append_mode append_filedrfilesrrs r;r r s@ B}0TUU"*&1# ([ R    b L :E2L --/CxCdClC SVZ:6:!Tkdtcz !))#r2::3C K{KlK  xQR LA KK5KFKE L   '    s1vuQx:6:; &$F8  69#ab'596M Nda\!Q )& ) N   !!VN  KLU+UXZ    !"";/;"";/9<''0k2''5'9'9+'F&GI   ^%7%7 %D%L*3N; 'LDLL&;N;<< C Os4G4 )NNTFN)FNNNNr) Fzutf-8wtNNTNNNN)F __future__rrcollections.abcr functoolsriorwarningsrrr dask._task_specr r ImportErrornumpyr+rr(fsspec.compressionr fsspec.corer r rr fsspec.utilsrpandas.api.typesrrrrrdask.dataframe dataframer dask.bytesr dask.corerdask.dataframe.backendsrdask.dataframe.utilsr dask.delayedr dask.utilsrrr<r"rrrrrrrrrrregister_inplacerrr rr rr=r;rFs" #77/$*)"*!?7 .    1hXB   nbDN . % #$   ZzHVD  7 & 7 7   bkk:u- /  k B r{{J >&*&    $Je FsD88EE