U /e @szddlmZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z mZmZmZddlZddlZddlmZdd lmZmZdd lmZmZdd lmZdd lm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'dd l(m)Z)ddl*m+Z+ddl,m-Z-m.Z.m/Z/m0Z0ddl1m2Z2m3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9m:Z:m;Z;e rJddldIddZ?edJdddddddd d!Z@edKd"ddddd#dd$d!Z@dLd%ddddd&dd'd!Z@eAd(d)ddej fd*d+ZBdMd,d-ZCdNd.d/ZDd0d1ZEdOd4d5ZFdPd6d7ZGd8d9ZHe.dQd;dd?d@ZIdRdAdBZJGdCdDdDe+ZKe.dddddddEdFdGZLeHjMe jH_MeGjMe jG_MdS)S) annotations)Iterable)partial)ceil)getitem)Lock) TYPE_CHECKINGrLiteraloverloadN)is_dask_collectiontokenize)BlockwiseDepDict blockwise)dataframe_creation_dispatch) DataFrameIndexSeries_concat_emulateapply_and_enforcehas_parallel_type new_dd_object)meta_lib_from_array)DataFrameIOFunction) check_metainsert_meta_param_descriptionis_series_like make_meta)Delayeddelayed)HighLevelGraph)DataFrameIOLayer)Mfuncname is_arraylikecsjdkrtdjf|dk r:t|ts4td|j}|dkrNt}tj dddk r|dkrvt j j }nVt |rtdnBtfdd|Dstt|j j }td j d |j jfd d |D}njd krRt |s |dkr|jg|j |dSt|d krH|jt jgj d||dStdnt jd rltd|dkrjdkrt tjd ndg}n4t|jd krtdt|djd dj gt|}ddt||D}|j|||dS)z8Create empty DataFrame or Series which has correct dtypezEfrom_array does not input more than 2D array, got array with shape %rNz3'index' must be an instance of dask.dataframe.Indexnamesz+For a struct dtype, columns must be a list.c3s|]}|jjkVqdSN)dtyper&).0ix8/tmp/pip-unpacked-wheel-dbjnr7gq/dask/dataframe/io/io.py Fsz#_meta_from_array..zdtype z doesn't have fields cs$g|]}|kr|dndqS)rZf8r-)r)n)fieldsr-r. Jsz$_meta_from_array..)namer(indexr()columnsr5z?For a 1d array, columns must be a scalar or single element listz Shape along axis 1 must be knownrz:Number of column names must match width of the array. Got z names for z columnscSs i|]\}}|tjg|dqS)r6)nparray)r)cdtr-r-r. csz$_meta_from_array..)ndim ValueErrorshape isinstancer_metarrgetattrr(listr&r8Zisscalarallsortedset differencer1Z_constructor_slicedlenZ _constructorr9isnanrangezip)r,r7r5metaextraZdtypesdatar-)r1r,r._meta_from_array/sd       &rOPc Cst|tjrt|||dSt|||d}ttdt||}|t|df}t|||}d|}i}tdt t t||D]b}t |t |||d|f} t |rt|| d|j|jf|||f<qt|| d|jf|||f<qt||||S)a|Read any sliceable array into a Dask Dataframe Uses getitem syntax to pull slices out of the array. The array need not be a NumPy array but must support slicing syntax x[50000:100000] and have 2 dimensions: x.ndim == 2 or have a record dtype: x.dtype == [('name', 'O'), ('balance', 'i8')] Parameters ---------- x : array_like chunksize : int, optional The number of rows per partition to use. columns : list or string, optional list of column names if DataFrame, single string if Series meta : object, optional An optional `meta` parameter can be passed for dask to specify the concrete dataframe type to use for partitions of the Dask dataframe. By default, pandas DataFrame is used. Returns ------- dask.DataFrame or dask.Series A dask DataFrame/Series )r7rLrLrr3z from_array-N)r@daArrayfrom_dask_arrayrOtuplerJrHr intrrslicertyper(r4r7r) r, chunksizer7rL divisionstokenr4dskr*rNr-r-r. from_arraygs!   r]Tz pd.DataFramez int | Noneboolz str | Noner)rN npartitionsrYsortr4returncCsdSr'r-rNr_rYr`r4r-r-r. from_pandassrcz pd.SeriesrcCsdSr'r-rbr-r-r.rcszpd.DataFrame | pd.SerieszDataFrame | Seriesc sttddtjrtdts,td|dk|dkkrDtdt}|dkrht|t sztdnt|t sztdpdt |||st d fiddgSj r̈j std |rj jsjd d tj ||d \}}nR|dkr$t|t stt t||}ttd ||tg}dgt|}fddtt|dd|ddD}t ||S)a{ Construct a Dask DataFrame from a Pandas DataFrame This splits an in-memory Pandas dataframe into several parts and constructs a dask.dataframe from those parts on which Dask.dataframe can operate in parallel. By default, the input dataframe will be sorted by the index to produce cleanly-divided partitions (with known divisions). To preserve the input ordering, make sure the input index is monotonically-increasing. The ``sort=False`` option will also avoid reordering, but will not result in known divisions. Note that, despite parallelism, Dask.dataframe may not always be faster than Pandas. We recommend that you stay with Pandas for as long as possible before switching to Dask.dataframe. Parameters ---------- data : pandas.DataFrame or pandas.Series The DataFrame/Series with which to construct a Dask DataFrame/Series npartitions : int, optional The number of partitions of the index to create. Note that if there are duplicate values or insufficient elements in ``data.index``, the output may have fewer partitions than requested. chunksize : int, optional The desired number of rows per index partition to use. Note that depending on the size and index of the dataframe, actual partition sizes may vary. sort: bool Sort the input by index first to obtain cleanly divided partitions (with known divisions). If False, the input will not be sorted, and all divisions will be set to None. Default is True. name: string, optional An optional keyname for the dataframe. Defaults to hashing the input Returns ------- dask.DataFrame or dask.Series A dask DataFrame/Series partitioned along the index Examples -------- >>> from dask.dataframe import from_pandas >>> df = pd.DataFrame(dict(a=list('aabbcc'), b=list(range(6))), ... index=pd.date_range(start='20100101', periods=6)) >>> ddf = from_pandas(df, npartitions=3) >>> ddf.divisions # doctest: +NORMALIZE_WHITESPACE (Timestamp('2010-01-01 00:00:00', freq='D'), Timestamp('2010-01-03 00:00:00', freq='D'), Timestamp('2010-01-05 00:00:00', freq='D'), Timestamp('2010-01-06 00:00:00', freq='D')) >>> ddf = from_pandas(df.a, npartitions=3) # Works with Series too! >>> ddf.divisions # doctest: +NORMALIZE_WHITESPACE (Timestamp('2010-01-01 00:00:00', freq='D'), Timestamp('2010-01-03 00:00:00', freq='D'), Timestamp('2010-01-05 00:00:00', freq='D'), Timestamp('2010-01-06 00:00:00', freq='D')) Raises ------ TypeError If something other than a ``pandas.DataFrame`` or ``pandas.Series`` is passed in. See Also -------- from_array : Construct a dask.DataFrame from an array that has record dtype read_csv : Construct a dask.DataFrame from a CSV file r5Nz,Dask does not support MultiIndex Dataframes.z+Input must be a pandas DataFrame or Series.;Exactly one of npartitions and chunksize must be specified.zSPlease provide npartitions as an int, or possibly as None if you specify chunksize.zSPlease provide chunksize as an int, or possibly as None if you specify npartitions.z from_pandas-rzIndex in passed data is non-numeric and contains nulls, which Dask does not entirely support. Consider passing `data.loc[~data.isna()]` instead.T)Z ascending)r_rYcs(i|] \}\}}|fj||qSr-)Ziloc)r)r*startstoprNr4r-r.r<.s zfrom_pandas..r3)r@rBpdZ MultiIndexNotImplementedErrorr TypeErrorr>rHrVr rr5ZisnaanyZ is_numericZis_monotonic_increasingZ sort_indexsorted_division_locationsAssertionErrorrrCrJ enumeraterK) rNr_rYr`r4ZnrowsrZ locationsr\r-rgr.rcsPK      pandasr7cCs<dd|D}|r&td|dt||||||S)a Construct a Dask DataFrame from a Python Dictionary Parameters ---------- data : dict Of the form {field : array-like} or {field : dict}. npartitions : int The number of partitions of the index to create. Note that depending on the size and index of the dataframe, the output may have fewer partitions than requested. orient : {'columns', 'index', 'tight'}, default 'columns' The "orientation" of the data. If the keys of the passed dict should be the columns of the resulting DataFrame, pass 'columns' (default). Otherwise if the keys should be rows, pass 'index'. If 'tight', assume a dict with keys ['index', 'columns', 'data', 'index_names', 'column_names']. dtype: bool Data type to force, otherwise infer. columns: string, optional Column labels to use when ``orient='index'``. Raises a ValueError if used with ``orient='columns'`` or ``orient='tight'``. constructor: class, default pd.DataFrame Class with which ``from_dict`` should be called with. Examples -------- >>> import dask.dataframe as dd >>> ddf = dd.from_dict({"num1": [1, 2, 3, 4], "num2": [7, 8, 9, 10]}, npartitions=2) cSsh|]}t|rt|qSr-)r rX)r)vr-r-r. ]szfrom_dict..zPfrom_dict doesn't currently support Dask collections as inputs. Objects of type z were given in the input dict.)valuesrjrc from_dict)rNr_orientr(r7 constructorZcollection_typesr-r-r.ru5s( rucKs(t|trtj|}||fd|i|S)aCreate a Dask partition for either a DataFrame or Series. Designed to be used with :func:`dask.blockwise.blockwise`. ``data`` is the array from which the partition will be created. ``index`` can be: 1. ``None``, in which case each partition has an independent RangeIndex 2. a `tuple` with two elements, the start and stop values for a RangeIndex for this partition, which gives a continuously varying RangeIndex over the whole Dask DataFrame 3. an instance of a ``pandas.Index`` or a subclass thereof The ``kwargs`` _must_ contain an ``initializer`` key which is set by calling ``type(meta)``. r5)r@rUriZ RangeIndex)rNr5 initializerkwargsr-r-r._partition_from_arrayjs  rzcCst||||d}dt||}|g}|j|jdkr6dndg}|j|ji}|dk r|j|jdkr|d|j|jd}t||j} | || |j dg|jf||j <nt t|jrdgt|jdd } nt|jd} dg} d} i} t|jdD]@\} }| |7} | | | f| | f<| | kr4| d 8} | | q| t| d dgt|rv|j|jt|d }n|jt|d }tt|df||d d|}tj|||d}t|||| S)aCreate a Dask DataFrame from a Dask Array. Converts a 2d array into a DataFrame and a 1d array into a Series. Parameters ---------- x : da.Array columns : list or string list of column names if DataFrame, single string if Series index : dask.dataframe.Index, optional An optional *dask* Index to use for the output Series or DataFrame. The default output index depends on whether `x` has any unknown chunks. If there are any unknown chunks, the output has ``None`` for all the divisions (one per chunk). If all the chunks are known, a default index with known divisions is created. Specifying `index` can be useful if you're conforming a Dask Array to an existing dask Series or DataFrame, and you would like the indices to match. meta : object, optional An optional `meta` parameter can be passed for dask to specify the concrete dataframe type to be returned. By default, pandas DataFrame is used. Examples -------- >>> import dask.array as da >>> import dask.dataframe as dd >>> x = da.ones((4, 2), chunks=(2, 2)) >>> df = dd.io.from_dask_array(x, columns=['a', 'b']) >>> df.compute() a b 0 1.0 1.0 1 1.0 1.0 2 1.0 1.0 3 1.0 1.0 See Also -------- dask.bag.to_dataframe: from dask.bag dask.dataframe._Frame.values: Reverse conversion dask.dataframe._Frame.to_records: Reverse conversion rQzfrom-dask-array-r%Zijr*Nrz@The index and array have different numbers of blocks. ({} != {})r3)mapping)r(r4rx)r7rxT) numblocksZ concatenateZ dependencies)rOr r4r=r|r_formatr>rZappendextend_namer8rIsumr?rHchunksror rr(rXr7rrzr from_collectionsr)r,r7r5rLr4Zgraph_dependenciesZarrays_and_indicesr|msgrZZ n_elementsrfZ index_mappingr* incrementryZblkgraphr-r-r.rT~s^-     rTcCsdS)zA dummy function to link results together in a graph We use this to enforce an artificial sequential ordering on tasks that don't explicitly pass around a shared resource Nr-)r[resultr-r-r._linksrFrUcCst|tjr^|dkr(ttt||S|dkr|rPddt|d|j DS|jddSnDt|tj r|dkr|rt| St|S|dkr| jddSdS)NrUdictcSsg|]\}}d|i|qS)r5r-)r)rtidxr-r-r.r2sz_df_to_bag..records)rv) r@rirrCmaprUZ itertuplesrKZto_dictr5ritemsZto_frame)dfr5r~r-r-r. _df_to_bags  rcsddlm}t|ttfs"tddt|dkrH|j}|jn8fddt | D}| | | | |||jS)a1Create Dask Bag from a Dask DataFrame Parameters ---------- index : bool, optional If True, the elements are tuples of ``(index, value)``, otherwise they're just the ``value``. Default is False. format : {"tuple", "dict", "frame"}, optional Whether to return a bag of tuples, dictionaries, or dataframe-like objects. Default is "tuple". If "frame", the original partitions of ``df`` will not be transformed in any way. Examples -------- >>> bag = df.to_bag() # doctest: +SKIP r)Bagz%df must be either DataFrame or Serieszto_bag-framecs"i|]\}}|ft|fqSr-)r)r)r*blockr~r5r4r-r.r<(szto_bag..)Z dask.bag.corerr@rrrkr ZdaskrroZ __dask_keys__updateZ__dask_optimize__Z__dask_graph__r_)rr5r~rr\r-rr.to_bag s  rcCs |tjS)azCreate Dask Array from a Dask Dataframe Warning: This creates a dask.array without precise shape information. Operations that depend on shape information, like slicing or reshaping, will not work. Examples -------- >>> df.to_records() # doctest: +SKIP See Also -------- dask.dataframe._Frame.values dask.dataframe.from_dask_array )Zmap_partitionsr" to_records)rr-r-r.r0sr from-delayedzEDelayed | distributed.Future | Iterable[Delayed | distributed.Future]z tuple | Literal['sorted'] | Nonestr)dfsrZprefix verify_metarac sfddlmt|s t|dr&|g}fdd|D}|D] }t|s.zExpected Delayed object, got %sNrEr3z+divisions should be a tuple of len(dfs) + 1-cSsi|]\}}|f|jqSr-r)r)r*inpr-r-r.r<sz from_delayed..T)Z produces_keys from_delayed)rLr#cSs|Sr'r-r+r-r-r.zfrom_delayed..)r4r7inputsio_func)compute_and_set_divisions) dask.delayedrr@rrkrX__name__rrZcomputerHrCr>r r!r rorrrr rZdask.dataframe.shuffler) rrLrZrritemZdivsr4layerrrr-rr.rCsR       rcs^|dkdkkrtdt|dr*|nt|}t|t|k}d}|rt|drd|j|ddnt|j|dd}|ot||k}nd}}dd}|rt||t||d }fd d }|dg} dg} |d} d} d} |r|t| nd}| t|kr:|| }|r| dkr>t||| kdd} |rvt|| }||krv| ||8} || } || }t|| }n| }|| d kr|r| d 7} | t|krt|| nt|} n| d 7} q|r| || d |t| d } |r|d 8}|t d |t| | } | || |d} q| |d | t|| | fS)aiFind division locations and values in sorted list Examples -------- >>> L = ['A', 'B', 'C', 'D', 'E', 'F'] >>> sorted_division_locations(L, chunksize=2) (['A', 'C', 'E', 'F'], [0, 2, 4, 6]) >>> sorted_division_locations(L, chunksize=3) (['A', 'D', 'F'], [0, 3, 6]) >>> L = ['A', 'A', 'A', 'A', 'B', 'B', 'B', 'C'] >>> sorted_division_locations(L, chunksize=3) (['A', 'B', 'C', 'C'], [0, 4, 7, 8]) >>> sorted_division_locations(L, chunksize=2) (['A', 'B', 'C', 'C'], [0, 4, 7, 8]) >>> sorted_division_locations(['A'], chunksize=2) (['A', 'A'], [0, 1]) NrduniqueF searchsortedleft)ZsiderTcst|kSr')rV)indrYZresidualr-r. chunksizessz-sorted_division_locations..chunksizesrhr3) r>rrr8rHrr9rVZnonzeromaxr)seqr_rYZ seq_unique duplicatesZ enforce_exactoffsetsZsubtract_driftrrZrpr*rZdriftZ divs_remaindivZ offs_remainposr-rr.rmsl       $    rmc@s6eZdZdZd ddZeddZdd Zd d ZdS) _PackedArgCallableaPacked-argument wrapper for DataFrameIOFunction This is a private helper class for ``from_map``. This class ensures that packed positional arguments will be expanded before the underlying function (``func``) is called. This class also handles optional metadata enforcement and column projection (when ``func`` satisfies the ``DataFrameIOFunction`` protocol). NFcCs6||_||_||_||_||_||_t|jt|_dSr') funcargsryrLenforce_metadatapackedr@ris_dataframe_io_func)selfrrryrLrrr-r-r.__init__s z_PackedArgCallable.__init__cCs|jr|jjSdSr')rrr7)rr-r-r.r7'sz_PackedArgCallable.columnscCs4|jr0t|j||j|j|j||j|jdS|S)N)rryrLrr) rrrproject_columnsrryrLrr)rr7r-r-r.r-s z"_PackedArgCallable.project_columnscCsV|js |g}|jr:t||jpg|j|jd|jp4iS|j||jpHg|jpRiS)N)Z_funcrA)rrrrrrLry)rZ packed_argr-r-r.__call__9s$z_PackedArgCallable.__call__)NNNFF) r __module__ __qualname____doc__rpropertyr7rrr-r-r-r.r s    r)rrLrZlabelr[rc Ost|stdt} t|}t|D]n\} } t| tsJtdt| z| t | Wq&t t fk rt| || <| t || Yq&Xq&t | dkrtdnt | dkrtd| dhkrtd| dd } | d d } | st |dkr t |dkrtd |d}d }ntt |}d }|p>> import pandas as pd >>> import dask.dataframe as dd >>> func = lambda x, size=0: pd.Series([x] * size) >>> inputs = ["A", "B"] >>> dd.from_map(func, inputs, size=2).compute() 0 A 1 A 0 B 1 B dtype: object This API can also be used as an alternative to other file-based IO functions, like ``read_parquet`` (which are already just ``from_map`` wrapper functions): >>> import pandas as pd >>> import dask.dataframe as dd >>> paths = ["0.parquet", "1.parquet", "2.parquet"] >>> dd.from_map(pd.read_parquet, paths).head() # doctest: +SKIP name timestamp 2000-01-01 00:00:00 Laura 2000-01-01 00:00:01 Oliver 2000-01-01 00:00:02 Alice 2000-01-01 00:00:03 Victor 2000-01-01 00:00:04 Bob Since ``from_map`` allows you to map an arbitrary function to any number of iterable objects, it can be a very convenient means of implementing functionality that may be missing from from other DataFrame-creation methods. For example, if you happen to have apriori knowledge about the number of rows in each of the files in a dataset, you can generate a DataFrame collection with a global RangeIndex: >>> import pandas as pd >>> import numpy as np >>> import dask.dataframe as dd >>> paths = ["0.parquet", "1.parquet", "2.parquet"] >>> file_sizes = [86400, 86400, 86400] >>> def func(path, row_offset): ... # Read parquet file and set RangeIndex offset ... df = pd.read_parquet(path) ... return df.set_index( ... pd.RangeIndex(row_offset, row_offset+len(df)) ... ) >>> def get_ddf(paths, file_sizes): ... offsets = [0] + list(np.cumsum(file_sizes)) ... return dd.from_map( ... func, paths, offsets[:-1], divisions=offsets ... ) >>> ddf = get_ddf(paths, file_sizes) # doctest: +SKIP >>> ddf.index # doctest: +SKIP Dask Index Structure: npartitions=3 0 int64 86400 ... 172800 ... 259200 ... dtype: int64 Dask Name: myfunc, 6 tasks See Also -------- dask.dataframe.from_delayed dask.layers.DataFrameIOLayer z"`func` argument must be `callable`z2All elements of `iterables` must be Iterable, got rz/`from_map` requires at least one Iterable inputr3z)All `iterables` must have the same lengthz+All `iterables` must have a non-zero lengthproduces_tasksF creation_infoNz9Multiple iterables not supported when produces_tasks=TrueTrZudfzMeta is not valid, `from_map` expects output to be a pandas object. Try passing a pandas object as meta or a dict or tuple representing the (name, dtype) of the columns.)rryrLrr)rrrr})callabler>rFrCror@rrXaddrHAttributeErrorrkpoprKr#r rr7rrrr$r?rrr!r rr)rrrLrZrr[r iterablesrylengthsr*iterablerrrrr4Zcolumn_projectionZmeta_is_emulatedrrrr-r-r.from_mapKs|              r)NNN)rPNN)NNTN)NNTN)NNTN)NN)NNN)FrU)FrU)NNrT)NN)N __future__rcollections.abcr functoolsrmathroperatorr threadingrtypingrr r Znumpyr8rqriZ dask.arrayr9rRZ dask.baser r Zdask.blockwiser rZdask.dataframe.backendsrZdask.dataframe.corerrrrrrrrZdask.dataframe.dispatchrZdask.dataframe.io.utilsrZdask.dataframe.utilsrrrrrrrZdask.highlevelgraphr Z dask.layersr!Z dask.utilsr"r#r$Z distributedlockrOr]rcZregister_inplacerurzrTrrrrrrmrrrr-r-r-r.s        (     8 5   4  p  &Q w?k