bi} ddlmZddlZddlZddlZddlZddlmZmZm Z m Z ddlm Z ddl m Z mZddlmZmZmZddlZddlZddlZddlmZddlmZdd lmZdd lm Z m!Z!m"Z"dd lm#Z$dd lm%Z%dd l&m'Z'ddlm(Z)ddl*m+Z+ddl,m-Z.ddl/Z0ddl1m2cm3Z3ddl0m4Z4m5Z5ddl6m7Z7ddl8m9Z9ddl:m;Z;mm?Z?m@Z@mAZAddlBmCZCddlDmEZEmFZFmGZGddlHmIZIddlJmKZKmLZLmMZMmNZNmOZOmPZPmQZQmRZRmSZSddlTmUZVddl/mWZWddlXmYZYmZZZm[Z[ddl\m]Z]ddl^m_Z_ddl`maZambZbdd lcmdZdmeZemfZfmgZgmhZhmiZimjZjmkZkmlZlmmZmmnZnmoZodd!lpmqZqmrZrdd"lsmtZtdd#lumvZvdd$lwmxZxmyZymzZzm{Z{m|Z|m}Z}m~Z~mZmZmZmZmZmZmZmZmZdd%lmZmZdd&lmZmZmZmZdd'lmZdd(lmZmZmZmZmZmZmZmZmZdd)lmZmZdd*lmZmZmZmZdd+lmZdd,lmZmZmZmZm#Z#mZmZmZdd-lmZmZdd.lmZdd/lmZmZmZmZmZmZmZmZmZmZmZmZdd0lmZdd1d2Zdhd3Zd4Zdhd5Zd6aGd7d8e?Zd9Zd:D]Zeeeejeeɬ;!deǫZGd?d@eǫZdAD]DZeeeΫrJeΫeeeeee̫eeeΫrJeΫeeeeeeͫFGdBdCeͫZGdDdEeǫZdidFZdjdGZdkdHZdIZeWjdJdKddejfdLZdldmdMZeWjdJ dndOZ dodPZ dpdQZe ej dqdRZe ej drdSZdeodddNdTdUZdsdVZdtdWZeedXgYdudZZe ejdhd[Ze ejdvd\Zd]Zeeod6d6dNdNddd^d_Zeeod6d6dNdNd`daZdbZeodd6dcddZdeZ ds dwdfZejeǫdgZy)x) annotationsN)CallableHashableIterableMapping)wraps)IntegralNumber)AnyClassVarLiteral)stringify_path)parse)CategoricalDtype) is_bool_dtypeis_datetime64_any_dtypeis_numeric_dtype) is_scalar)is_timedelta64_dtype)is_extension_array_dtype)fsfirst)computeget_annotationsnew_collection)OptimizerStage)DictTaskRef)Array)DaskMethodsMixinis_dask_collectionnamed_schedulers)flatten) PANDAS_GE_210 PANDAS_GE_220PANDAS_VERSION)CachedAccessor) _concat_convert_to_numeric_repr_data_series_sqrt_and_convert_to_timedeltacheck_divisionshas_parallel_typeis_dataframe_likeis_series_like meta_warning_exprdataframe_creation_dispatch)CategoricalAccessor Categorize GetCategories)Concat)DatetimeAccessor)DescribeNonNumericDescribeNumeric) BFillDiffEvalFFill FillnaCheckQueryShift ToDatetime ToNumeric ToTimedelta _DelayedExpr no_default) JoinRecursiveMerge)SeriesQuantile)RepartitionQuantiles)CorrCovCustomReductionDropDuplicates IndexCountIsMonotonicDecreasingIsMonotonicIncreasingLenMemoryUsageFrameMemoryUsageIndexMomentNLargest NSmallest PivotTableUnique ValueCounts) RepartitionRepartitionFreq)RearrangeByColumnSetIndexSetIndexBlockwise SortValues)StringAccessor) PANDAS_GE_300 _BackendData_convert_to_list_get_shuffle_preferring_order_is_any_real_numeric_dtype_maybe_from_pandas_raise_if_object_series_validate_axisget_specified_shuffle)FromPandasDivisions FromScalars)get_parallel_typeis_categorical_dtype make_meta meta_nonempty)warn_dtype_mismatch)AttributeNotImplementedErrorhas_known_categories index_summaryinsert_meta_param_descriptionrmeta_frame_constructormeta_series_constructorpyarrow_strings_enabled)Delayeddelayed)_tokenize_deterministic) IndexCallableM derived_fromget_default_shuffle_methodget_meta_library is_arraylike key_splitmaybe_pluralize memory_repr put_linesrandom_state_datatypename) get_templatewrap_apic |J||Dcgc] }t|tr |jn|"c}i|}t|tjr t |S|Scc}wN) isinstance FrameBaseexprExprr)rargskwargsargresults _/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/dask/dataframe/dask_expr/_collection.py_wrap_expr_apirsf    EI Jcji0#((c 9 J F&$))$f%% M Ks%Ac|Jt|tr |j}nrt|tjrXt ||j |j}|jdk(r't|jr||jd}t|tjs t|s t|r|j|j}t|tjs%tt|j||Stj |j|r%tt|j||Sttj"|||S)Nindexcolumnsr)rrrdar!from_dask_arrayrrndimlenrr0r1_create_alignable_framergetattrare_co_alignedOpAlignPartitions)selfotherops r _wrap_expr_oprs >>%#  E288 $TZZN 99>c$,,/$,,q/*E udii ( e $ % ,,U388 eTYY '4gdii4U;<<  TYY .4gdii4U;<<d44T5"EFFcr|tk(rdfd }n|tk(rdfd }ntd||_|S)zz Add method operators to Series or DataFrame like DataFrame.add. _wrap_expr_method_operator("add", DataFrame) c | tdt|}t|st|r |dvrt |s|j |}|dvr!t |trdd}t||}t |trNtj|j|js$ttj|||||Sttj|||||S)Nlevel must be None)rr)rrz Unable to z dd.Series with axis=1rframeraxislevel fill_value)nameleftrightrrr)NotImplementedErrorrlr0r1r#rrSeries ValueErrorrrrrMethodOperatorAlignMethodOperator)rrrrrmsgrrs rmethodz*_wrap_expr_method_operator..methods )*>??!$'D"%(!%(L((/44U;~%eV,&tf,BCC$S/)E%+D4G4G 5::5&,,##!##-   "##)   rc | tdt|}t|rt|s|j |}|}t |t rNtj|j|js$ttj|||||Sttj|||||S)Nrr)rrrrrr) rrlr1r#rrrrrrrr)rrrrrrrs rrz*_wrap_expr_method_operator..methods )*>??!$'De$-?-F44U;E%+D4G4G 5::5&,,##!##-   "##)   rz)Cannot create method operator for class_=)rNNNNr) DataFramerr__name__)rclass_rs` r_wrap_expr_method_operatorrsD * X 6 ! H"$Nvi"PQQFO MrcR|Jtt|j|Sr)rrr)rrs r_wrap_unary_expr_oprs( >> 0'$))R02 33rTc  eZdZdZeej dedZedZdZ e ddZ e dZ e jdZe d Ze d Ze d Ze d Ze d Zej*dZdZdZe dZdZdZdZdZdZddZddZdddZ dddZ!dZ"e dZ#dZ$d Z%d!Z&d"Z'ddd#Z(d$Z)ed%Z*d&Z+d'Z,ddfd) Z-e d*Z.e d+Z/e/j*d,Z/ddd-Z0ddd/Z1ddd0Z2ddd1Z3e4e5jld2Z7d3Z8e d4Z9d5Z:e;d(ddd(d(f dd6Z<e4e5jldd7Z=d8Z>e?d9:e;ddd(d(ddd;d<Z@e?d9:e;ddd(d(d=d>ZA d dd?ZB d dd@ZCe dAZDdBZEdCZFdDZGe4e5jl ddFZHdGZIe4e5jl ddHZJeJZKe4e5jl ddIZLe4e5jl ddJZMdKZNe4e5jl ddLZOe4e5jl ddMZPePZQe4e5jl ddNZRdOZS ddPZT ddQZUe4e5jl ddRZVe4e5jlddSZWe4e5jlddTZXe4e5jlddUZYe4e5jlddVZZe4e5jlddWZ[e4e5jlddXZ\e4e5jlddYZ]e4e5jldZZ^e4e5jld[Z_e4e5jld\Z`e4e5jldd]Zae4e5jld^Zbdd_Zce4e5jldd`Zde4e5jleejfdaZge4e5jleejfdbZhe4e5jlde;d(fdcZie4e5jldddZje4e5jlddeZke4e5jlddfZle4e5jlddgZme4e5jlddhZne4e5jle;e;e;dEfdiZoddjZpe4e5jlddkZqddlZre4e5jlddmZse4e5jlddnZte4e5jlddoZue4e5jlddpZvdde;dddddfdqZwdddrZxe dsZye4e5jldtZze4e5jlduZ{dddvZ|e}d.dwdddxdyZ~dzZ d dd{Zd|Zd}Zdd~ZdZddZddZddZdddZe4e5je;fdZe4e5jldZddZdZe dZxZS)rz&Base class for Expr-backed Collectionsthreadssyncc |Sr)dskkeysrs rzFrameBase.0srcltr'tx}rdatjd|d||_y)NFzDask annotations zA detected. Annotations will be ignored when using query-planning.)_WARN_ANNOTATIONSrwarningswarnr4)rrannots r__init__zFrameBase.__init__2s: ?+<"<%"< %  MM#E7*kl  rc|jSrr3rs rrzFrameBase.expr;s zzrc.|jjSr)r_metars rrzFrameBase._meta?syyrc,t|jSr)rsrrs r_meta_nonemptyzFrameBase._meta_nonemptyCsTZZ((rc.|jjS)a Tuple of ``npartitions + 1`` values, in ascending order, marking the lower/upper bounds of each partition's index. Divisions allow Dask to know which partition will contain a given value, significantly speeding up operations like `loc`, `merge`, and `groupby` by not having to search the full dataset. Example: for ``divisions = (0, 10, 50, 100)``, there are three partitions, where the index in each partition contains values [0, 10), [10, 50), and [50, 100], respectively. Dask therefore knows ``df.loc[45]`` will be in the second partition. When every item in ``divisions`` is ``None``, the divisions are unknown. Most operations can still be performed, but some will be much slower, and a few may fail. It is not supported to set ``divisions`` directly. Instead, use ``set_index``, which sorts and splits the data as needed. See https://docs.dask.org/en/latest/dataframe-design.html#partitions. )r divisionsrs rrzFrameBase.divisionsGs,yy"""rc.|jjS)zReturn number of partitions)r npartitionsrs rrzFrameBase.npartitions_syy$$$rcB|jjjS)zReturn data types)rrdtypesrs rrzFrameBase.dtypesdsyy%%%rc@t|jjS)zSize of the Series or DataFrame as a Delayed object. Examples -------- >>> series.size # doctest: +SKIP )rrsizers rrzFrameBase.sizeisdiinn--rc.|jjSr)rrrs rrzFrameBase.columnstszz!!!rct|t|jk7r||jj_t j |||_yr)rrr4rr ColumnsSetter)rrs rrzFrameBase.columnsxs= w<3t||, ,'.DJJ   $''g6 rc>ttj|S)zhForget division information. This is useful if the divisions are no longer meaningful. )rrClearDivisionsrs rclear_divisionszFrameBase.clear_divisionss d11$788rcFtt|jSr)rrUrrs r__len__zFrameBase.__len__sc$i(0022rctd)Nz&nbytes is not implemented on DataFramerrs rnbyteszFrameBase.nbytess!"JKKrc(t|jffSr)rr4rs r __reduce__zFrameBase.__reduce__s },,rct|trwtj|j|jst tj ||St |jj |jSt|trddlm }td|j|j|jfD}|jdk(r3|r1||jj rt"r|j$|S|j&|St|t(j*s t-|r t/|}nKt|t.r|j1}n*t|t(j2r|j5}t |jj |S)Nr)is_float_dtypec3<K|]}t|tywr)rr ).0is r z(FrameBase.__getitem__..s#,- 1h'#)rrrrr FilterAlign __getitem__slicepandas.api.typesranystartstepstoprrdtypereiloclocnpndarrayr1listcopygenericitem)rrris_integer_slices rrzFrameBase.__getitem__s9 eY '&&tyy%**=%d&6&6tU&CDD!$))"7"7 "CD D u % 7"#27++uzz5::1V#   Q$' (8(89]yy''xx& eRZZ (N5,AKE t $JJLE rzz *JJLEdii33E:;;rcXt|j|jjfSr)typerr4_namers r__dask_tokenize__zFrameBase.__dask_tokenize__s!Dz""DJJ$4$444rc |jjd}d}t|ts>t |j s)|j ddjdd}d|}t |jjDchc]}|jc}}|j|jj|t|jt|d |j Scc}w) Nmax_rowszFDask {klass} Structure: {data} Dask Name: {name}, {n_expr} Expr={expr} Index DivisionsEmpty expression)klassdatarn_exprr) _repr_data to_stringrrrr partitionreplacerwalkr format __class__rrr)rr_str_fmters r__repr__zFrameBase.__repr__s **A*6$'DLL0A>>$'+33G[IDz*Htyy~~'78!agg89..))4::&"6<8   9sC=cHtd|jjd)NzThe truth value of a z& is ambiguous. Use a.any() or a.all().)rr rrs r__bool__zFrameBase.__bool__s+#DNN$;$;#<=& &  rNc Htj|jSr)rarrayr)rrrs r __array__zFrameBase.__array__sxx ''rTc R|j|}tj|fi|SNfuse)optimizer"persist)rr,routs rr.zFrameBase.persists(mmm&''6v66rc|}t|ts|jd}|jj ||S)aOutputs statistics about every node in the expression. analyze optimizes the expression and triggers a computation. It records statistics like memory usage per partition to analyze how data flow through the graph. .. warning:: analyze adds plugins to the scheduler and the workers that have a non-trivial cost. This method should not be used in production workflows. Parameters ---------- filename: str, None File to store the graph representation. format: str, default is png File format for the graph representation. Returns ------- None, but writes a graph representation of the expression enriched with statistics to disk. rr)filenamer)rScalar repartitionranalyze)rr2rr/s rr5zFrameBase.analyzes=,#v&//a/0Cxx&AArc|}t|ts|jd}|jj ||S)a7Create a graph representation of the Expression. explain runs the optimizer and creates a graph of the optimized expression with graphviz. No computation is triggered. Parameters ---------- stage: {"logical", "simplified-logical", "tuned-logical", "physical", "simplified-physical", "fused"} The optimizer stage that is returned. Default is "fused". - logical: outputs the expression as is - simplified-logical: simplifies the expression which includes predicate pushdown and column projection. - tuned-logical: applies additional optimizations like partition squashing - physical: outputs the physical expression; this expression can actually be computed - simplified-physical: runs another simplification after the physical plan is generated - fused: fuses the physical expression to reduce the nodes in thr graph. .. warning:: The optimizer stages are subject to change. format: str, default None The format of the output. Default is "png". Returns ------- None, but opens a new window with the graph visualization and outputs a file with the graph representation. rr1)rr3r4rexplain)rstagerr/s rr7zFrameBase.explains:>#v&//a/0Cxxv..rc6|jjS)zOutputs a string representation of the DataFrame. The expression is returned as is. Please run optimize manually if necessary. Returns ------- None, the representation is put into stdout. )rpprintrs rr:zFrameBase.pprint syy!!rc"|jSr)__dask_graph__rs rdaskzFrameBase.dasks""$$rcZ|j}|j}|jSr)rlower_completelyr<rr/s rr<zFrameBase.__dask_graph__s(ii""$!!##rcZ|j}|j}|jSr)rr? __dask_keys__r@s rrBzFrameBase.__dask_keys__!s(ii""$  ""rcHt|jjSr)rrsimplifyrs rrDzFrameBase.simplify&sdii00233rcJt|jjiSr)rr lower_oncers rrFzFrameBase.lower_once)sdii222677rcLt|jj|S)a2Optimizes the DataFrame. Runs the optimizer with all steps over the DataFrame and wraps the result in a new DataFrame collection. Only use this method if you want to analyze the optimized expression. Parameters ---------- fuse: bool, default True Whether to fuse the expression tree after running the optimizer. It is often easier to look at the non-fused expression when analyzing the result. Returns ------- The optimized Dask Dataframe r+rrr-)rr,s rr-zFrameBase.optimize,s!$dii00d0;<;A*A>#A*$A>* A66A99A>Fc ^|rt|di|S|jjdi|S)zVisualize the expression or task graph Parameters ---------- tasks: Whether to visualize the task graph. By default the expression graph will be visualized instead. r)super visualizer)rtasksrr s rrbzFrameBase.visualizeis5 7$.v. ."tyy"",V,,rc.|jjS)zWhether the divisions are known. This check can be expensive if the division calculation is expensive. DataFrame.set_index is a good example where the calculation needs an inspection of the data. )rknown_divisionsrs rrezFrameBase.known_divisionsvsyy(((rc@t|jjS)zReturn dask Index instance)rrrrs rrzFrameBase.indexsdiioo..rctj|j|jsJdtj||}||_y)Nz(value needs to be aligned with the index)rr AssignIndexr4)rvaluer4s rrzFrameBase.indexsJ"" IIuzz  6 5 6   u- rc@ttj||S)a{Reset the index to the default index. Note that unlike in ``pandas``, the reset index for a Dask DataFrame will not be monotonically increasing from 0. Instead, it will restart at 0 for each partition (e.g. ``index1 = [0, ..., 10], index2 = [0, ...]``). This is due to the inability to statically know the full length of the index. For DataFrame with multi-level index, returns a new DataFrame with labeling information in the columns under the index names, defaulting to 'level_0', 'level_1', etc. if any are None. For a standard index, the index name will be used (if set), otherwise a default 'index' or 'level_0' (if 'index' is already taken) will be used. Parameters ---------- drop : boolean, default False Do not try to insert index into dataframe columns. )rr ResetIndex)rdrops r reset_indexzFrameBase.reset_indexs(doodD9::rrclttj|||}|r|j}|S)a0First n rows of the dataset Parameters ---------- n : int, optional The number of rows to return. Default is 5. npartitions : int, optional Elements are only taken from the first ``npartitions``, with a default of 1. If there are fewer than ``n`` rows in the first ``npartitions`` a warning will be raised and any found rows returned. Pass -1 to use all partitions. compute : bool, optional Whether to compute the result, default is True. )nr)rrHeadr)rrorrr/s rheadzFrameBase.heads.TYYtqkJK ++-C rcjttj||}|r|j}|S)zkLast n rows of the dataset Caveat, the only checks the last n rows of the last partition. )ro)rrTailr)rrorr/s rtailzFrameBase.tails, TYYtq12 ++-C rcJ|dur tdt|jS)aMake a copy of the dataframe This is strictly a shallow copy of the underlying computational graph. It does not affect the underlying data Parameters ---------- deep : boolean, default False The deep value must be `False` and it is declared as a parameter just for compatibility with third-party libraries like cuDF and pandas FzfThe `deep` value must be False. This is strictly a shallow copy of the underlying computational graph.)rrrrdeeps rrzFrameBase.copys/ u 9 dii((rclt|tr&ttjtjf}ntf}t||rnt|tr>|j dk(r/|j dk(r ttj||Stdtt|zt|tr tt|}t!d|Dsntj"j$j'|d}hd}||vr&t)j*|t,t/|}nt)j0|}ttj|tj2t5|d t7|z S#t$rYwxYw) Nr)valueszPassing a %r to `isin`c32K|]}t|ywr)r#)rvs rrz!FrameBase.isin..s=)!,=sFskipna> mixed-integer unknown-arraytimemixedperioddecimal categorical)rcountdelayed-r)rrrpdrrrrrIsinrrr rset TypeErrorrapitypes infer_dtyperfromiterrVrasarrayrHr}r~)rry bad_types inferred_type object_likes risinzFrameBase.isins[ dI &"BIIr||$v,AW&WX X fd #  c&k*=f== " 8 8 8 N  !K/[[vS[QFZZ/F II((F6Mf6U)UV   '  s F'' F32F3c0t|tr|g}tj|jt |j }t|jt|jsJttj||S)N)r) rintrarangerrVtolistrissubsetrangerr Partitionsrrs r _partitionszFrameBase._partitions sv eS !GE $**&9%@GGI5z""5)9)9#:;;;doodE:;;rc,t|jS)aSlice dataframe by partitions This allows partitionwise slicing of a Dask Dataframe. You can perform normal Numpy-style slicing, but now rather than slice elements of the array you slice along partitions so, for example, ``df.partitions[:5]`` produces a new Dask Dataframe of the first five partitions. Valid indexers are integers, sequences of integers, slices, or boolean masks. Examples -------- >>> df.partitions[0] # doctest: +SKIP >>> df.partitions[:3] # doctest: +SKIP >>> df.partitions[::10] # doctest: +SKIP Returns ------- A Dask DataFrame )rrrs r partitionszFrameBase.partitionss(T--..rcd|cxkr|jksnd|j}t||j|S)ap Get a dask DataFrame/Series representing the `nth` partition. Parameters ---------- n : int The 0-indexed partition number to select. Returns ------- Dask DataFrame or Series The same type as the original object. See Also -------- DataFrame.partitions rzn must be 0 <= n < )rrr)rrors r get_partitionzFrameBase.get_partition/sC$A(((('(8(8'9:CS/ !q!!rc  |tur |s td|tur |r td|xs |j}t|tr5t j |j |j stdtjjj|rt|s t|}nt|ttfr|g}n|rg}|Dcgc]+}||jvr||j j"k7r|-} }| rt%d| d|xs t'dk(rddlm} | |j,t/d |j,jDrL|j,jD cic]} t| tr| t1| !} } td | t3t5||||t7||| } |r| j9d | j, S| Scc}wcc} w)aCRearrange DataFrame into new partitions Uses hashing of `on` to map rows to output partitions. After this operation, rows with the same value of `on` will be in the same partition. Parameters ---------- on : str, list of str, or Series, Index, or DataFrame Column names to shuffle by. ignore_index : optional Whether to ignore the index. Default is ``False``. npartitions : optional Number of output partitions. The partition count will be preserved by default. shuffle_method : optional Desired shuffle method. Default chosen at optimization time. on_index : bool, default False Whether to shuffle on the index. Mutually exclusive with 'on'. Set this to ``True`` if 'on' is not provided. force : bool, default False This forces the optimizer to keep the shuffle even if the final expression could be further simplified. **options : optional Algorithm-specific options. Notes ----- This does not preserve a meaningful index/partitioning scheme. This is not deterministic if done in parallel. Examples -------- >>> df = df.shuffle(df.columns[0]) # doctest: +SKIP z~Must shuffle on either columns or the index; currently shuffling on neither. Pass column(s) to 'on' or set 'on_index' to True.zgCannot shuffle on both columns and the index. Do not pass column(s) to 'on' or set 'on_index' to False.zAindex must be aligned with the DataFrame to use as shuffle index.zCannot shuffle on z', column(s) not in dataframe to shufflep2pr)check_dtype_supportc3>K|]}t|t ywrrstrrcs rrz$FrameBase.shuffle..sFaz!S))Fz0p2p requires all column names to be str, found: ) index_shufflec|Srrxs rrz#FrameBase.shuffle..srrQ)rIrrrrrrrrr is_list_liker#rrrrrrKeyErrorrdistributed.shuffle._arrowrrrr rr`rmmap_partitions)ron ignore_indexrshuffle_methodon_indexforceoptions index_colbad_colsrr unsupportedress rshufflezFrameBase.shuffleFs Z  HM z !h6  "5T%5%5 b) $&&tyy"'':Wvv||((,5G5K"XBc +T"$T\\1 TZZ__8TH ( 2YZ  :8:u D F  +F4::3E3EFF(, (:(:#$*QPSBTAtAwJ  F{mT %n5&    %%k %B B O s"0H H "H c&ddlm}|||||S)Nr) Resampler)closedlabel)dask.dataframe.tseries.resampler)rrulerrrs rresamplezFrameBase.resamples=tF%@@rc "ddlm}|||fi|S)a2Provides rolling transformations. Parameters ---------- window : int, str, offset Size of the moving window. This is the number of observations used for calculating the statistic. When not using a ``DatetimeIndex``, the window size must not be so large as to span more than one adjacent partition. If using an offset or offset alias like '5D', the data must have a ``DatetimeIndex`` min_periods : int, default None Minimum number of observations in window required to have a value (otherwise result is NA). center : boolean, default False Set the labels at the center of the window. win_type : string, default None Provide a window type. The recognized window types are identical to pandas. axis : int, str, None, default 0 This parameter is deprecated with ``pandas>=2.1``. Returns ------- a Rolling object on which to call a method to compute a statistic r)Rolling)!dask.dataframe.dask_expr._rollingr)rwindowrrs rrollingzFrameBase.rollings4 >tV.v..r padrQenforce_metadatatransform_divisionsralign_dataframes parent_metarequired_columnsc 2t||g| |||||||d| S)aApply a Python function to each partition Parameters ---------- func : function Function applied to each partition. args, kwargs : Arguments and keywords to pass to the function. Arguments and keywords may contain ``FrameBase`` or regular python objects. DataFrame-like args (both dask and pandas) must have the same number of partitions as ``self`` or comprise a single partition. Key-word arguments, Single-partition arguments, and general python-object arguments will be broadcasted to all partitions. enforce_metadata : bool, default True Whether to enforce at runtime that the structure of the DataFrame produced by ``func`` actually matches the structure of ``meta``. This will rename and reorder columns for each partition, and will raise an error if this doesn't work, but it won't raise if dtypes don't match. transform_divisions : bool, default True Whether to apply the function onto the divisions and apply those transformed divisions to the output. clear_divisions : bool, default False Whether divisions should be cleared. If True, `transform_divisions` will be ignored. required_columns : list or None, default None List of columns that ``func`` requires for execution. These columns must belong to the first DataFrame argument (in ``args``). If None is specified (the default), the query optimizer will assume that all input columns are required. $META Examples -------- Given a DataFrame, Series, or Index, such as: >>> import pandas as pd >>> import dask.dataframe as dd >>> df = pd.DataFrame({'x': [1, 2, 3, 4, 5], ... 'y': [1., 2., 3., 4., 5.]}) >>> ddf = dd.from_pandas(df, npartitions=2) One can use ``map_partitions`` to apply a function on each partition. Extra arguments and keywords can optionally be provided, and will be passed to the function after the partition. Here we apply a function with arguments and keywords to a DataFrame, resulting in a Series: >>> def myadd(df, a, b=1): ... return df.x + df.y + a + b >>> res = ddf.map_partitions(myadd, 1, b=2) >>> res.dtype dtype('float64') Here we apply a function to a Series resulting in a Series: >>> res = ddf.x.map_partitions(lambda x: len(x)) # ddf.x is a Dask Series Structure >>> res.dtype dtype('int64') By default, dask tries to infer the output metadata by running your provided function on some fake data. This works well in many cases, but can sometimes be expensive, or even fail. To avoid this, you can manually specify the output metadata with the ``meta`` keyword. This can be specified in many forms, for more information see ``dask.dataframe.utils.make_meta``. Here we specify the output is a Series with no name, and dtype ``float64``: >>> res = ddf.map_partitions(myadd, 1, b=2, meta=(None, 'f8')) Here we map a function that takes in a DataFrame, and returns a DataFrame with a new column: >>> res = ddf.map_partitions(lambda df: df.assign(z=df.x * df.y)) >>> res.dtypes x int64 y float64 z float64 dtype: object As before, the output metadata can also be specified manually. This time we pass in a ``dict``, as the output is a DataFrame: >>> res = ddf.map_partitions(lambda df: df.assign(z=df.x * df.y), ... meta={'x': 'i8', 'y': 'f8', 'z': 'f8'}) In the case where the metadata doesn't change, you can also pass in the object itself directly: >>> res = ddf.map_partitions(lambda df: df.head(), meta=ddf) Also note that the index and divisions are assumed to remain unchanged. If the function you're mapping changes the index/divisions, you'll need to pass ``clear_divisions=True``. >>> ddf.map_partitions(func, clear_divisions=True) # doctest: +SKIP Your map function gets information about where it is in the dataframe by accepting a special ``partition_info`` keyword argument. >>> def func(partition, partition_info=None): ... pass This will receive the following information: >>> partition_info # doctest: +SKIP {'number': 1, 'division': 3} For each argument and keyword arguments that are dask dataframes you will receive the number (n) which represents the nth partition of the dataframe and the division (the first index value in the partition). If divisions are not known (for instance if the index is not sorted) then you will get None as the division. r)r) rfuncrQrrrrrrrrs rrzFrameBase.map_partitionssHF      - 3+-#-    rrQrrrrc 2t||||g| |||||d| S)a.Apply a function to each partition, sharing rows with adjacent partitions. This can be useful for implementing windowing functions such as ``df.rolling(...).mean()`` or ``df.diff()``. Parameters ---------- func : function Function applied to each partition. before : int, timedelta or string timedelta The rows to prepend to partition ``i`` from the end of partition ``i - 1``. after : int, timedelta or string timedelta The rows to append to partition ``i`` from the beginning of partition ``i + 1``. args, kwargs : Positional and keyword arguments to pass to the function. Positional arguments are computed on a per-partition basis, while keyword arguments are shared across all partitions. The partition itself will be the first positional argument, with all other arguments passed *after*. Arguments can be ``Scalar``, ``Delayed``, or regular Python objects. DataFrame-like args (both dask and pandas) will be repartitioned to align (if necessary) before applying the function; see ``align_dataframes`` to control this behavior. enforce_metadata : bool, default True Whether to enforce at runtime that the structure of the DataFrame produced by ``func`` actually matches the structure of ``meta``. This will rename and reorder columns for each partition, and will raise an error if this doesn't work, but it won't raise if dtypes don't match. transform_divisions : bool, default True Whether to apply the function onto the divisions and apply those transformed divisions to the output. align_dataframes : bool, default True Whether to repartition DataFrame- or Series-like args (both dask and pandas) so their divisions align before applying the function. This requires all inputs to have known divisions. Single-partition inputs will be split into multiple partitions. If False, all inputs must have either the same number of partitions or a single partition. Single-partition inputs will be broadcast to every partition of multi-partition inputs. $META Notes ----- Given positive integers ``before`` and ``after``, and a function ``func``, ``map_overlap`` does the following: 1. Prepend ``before`` rows to each partition ``i`` from the end of partition ``i - 1``. The first partition has no rows prepended. 2. Append ``after`` rows to each partition ``i`` from the beginning of partition ``i + 1``. The last partition has no rows appended. 3. Apply ``func`` to each partition, passing in any extra ``args`` and ``kwargs`` if provided. 4. Trim ``before`` rows from the beginning of all but the first partition. 5. Trim ``after`` rows from the end of all but the last partition. Examples -------- Given a DataFrame, Series, or Index, such as: >>> import pandas as pd >>> import dask.dataframe as dd >>> df = pd.DataFrame({'x': [1, 2, 4, 7, 11], ... 'y': [1., 2., 3., 4., 5.]}) >>> ddf = dd.from_pandas(df, npartitions=2) A rolling sum with a trailing moving window of size 2 can be computed by overlapping 2 rows before each partition, and then mapping calls to ``df.rolling(2).sum()``: >>> ddf.compute() x y 0 1 1.0 1 2 2.0 2 4 3.0 3 7 4.0 4 11 5.0 >>> ddf.map_overlap(lambda df: df.rolling(2).sum(), 2, 0).compute() x y 0 NaN NaN 1 3.0 3.0 2 6.0 5.0 3 11.0 7.0 4 18.0 9.0 The pandas ``diff`` method computes a discrete difference shifted by a number of periods (can be positive or negative). This can be implemented by mapping calls to ``df.diff`` to each partition after prepending/appending that many rows, depending on sign: >>> def diff(df, periods=1): ... before, after = (periods, 0) if periods > 0 else (0, -periods) ... return df.map_overlap(lambda df, periods=1: df.diff(periods), ... periods, 0, periods=periods) >>> diff(ddf, 1).compute() x y 0 NaN NaN 1 1.0 1.0 2 2.0 1.0 3 3.0 1.0 4 4.0 1.0 If you have a ``DatetimeIndex``, you can use a ``pd.Timedelta`` for time- based windows or any ``pd.Timedelta`` convertible string: >>> ts = pd.Series(range(10), index=pd.date_range('2017', periods=10)) >>> dts = dd.from_pandas(ts, npartitions=2) >>> dts.map_overlap(lambda df: df.rolling('2D').sum(), ... pd.Timedelta('2D'), 0).compute() 2017-01-01 0.0 2017-01-02 1.0 2017-01-03 3.0 2017-01-04 5.0 2017-01-05 7.0 2017-01-06 9.0 2017-01-07 11.0 2017-01-08 13.0 2017-01-09 15.0 2017-01-10 17.0 Freq: D, dtype: float64 r) map_overlap) rrbeforeafterrQrrrrrrs rrzFrameBase.map_overlapksH^        - 3+-    rc *t|du|du|du|dugdk7r td| t||Gt|jdt j s tdtt||Stt||||||S)a Repartition a collection Exactly one of `divisions`, `npartitions` or `partition_size` should be specified. A ``ValueError`` will be raised when that is not the case. Parameters ---------- divisions : list, optional The "dividing lines" used to split the dataframe into partitions. For ``divisions=[0, 10, 50, 100]``, there would be three output partitions, where the new index contained [0, 10), [10, 50), and [50, 100), respectively. See https://docs.dask.org/en/latest/dataframe-design.html#partitions. npartitions : int, Callable, optional Approximate number of partitions of output. The number of partitions used may be slightly lower than npartitions depending on data distribution, but will never be higher. The Callable gets the number of partitions of the input as an argument and should return an int. partition_size : str, optional Max number of bytes of memory for each partition. Use numbers or strings like 5MB. If specified npartitions and divisions will be ignored. Note that the size reflects the number of bytes used as computed by pandas.DataFrame.memory_usage, which will not necessarily match the size when storing to disk. .. warning:: This keyword argument triggers computation to determine the memory size of each partition, which may be expensive. force : bool, default False Allows the expansion of the existing divisions. If False then the new divisions' lower and upper bounds must be the same as the old divisions'. freq : str, pd.Timedelta A period on which to partition timeseries data like ``'7D'`` or ``'12h'`` or ``pd.Timedelta(hours=12)``. Assumes a datetime index. Notes ----- Exactly one of `divisions`, `npartitions`, `partition_size`, or `freq` should be specified. A ``ValueError`` will be raised when that is not the case. Also note that ``len(divisions)`` is equal to ``npartitions + 1``. This is because ``divisions`` represents the upper and lower bounds of each partition. The first item is the lower bound of the first partition, the second item is the lower bound of the second partition and the upper bound of the first partition, and so on. The second-to-last item is the lower bound of the last partition, and the last (extra) item is the upper bound of the last partition. Examples -------- >>> df = df.repartition(npartitions=10) # doctest: +SKIP >>> df = df.repartition(divisions=[0, 5, 10, 20]) # doctest: +SKIP >>> df = df.repartition(freq='7d') # doctest: +SKIP See Also -------- DataFrame.memory_usage_per_partition pandas.DataFrame.memory_usage NrzWPlease provide exactly one of the ``npartitions=`` or ``divisions=`` keyword arguments.rz0Can only repartition on frequency for timeseries) sumrr.rrr Timestamprrr_r^)rrrpartition_sizefreqrs rr4zFrameBase.repartitionsP T)t+"$.$   4   I &  dnnQ/> RSS!/$"=> >!D+y%QUV rc |dur,t|jtj}|j}|j ||}||_|||_|S)a Convert a dask DataFrame to a dask array. Parameters ---------- lengths : bool or Sequence of ints, optional How to determine the chunks sizes for the output array. By default, the output array will have unknown chunk lengths along the first axis, which can cause some later operations to fail. * True : immediately compute the length of each partition * Sequence : a sequence of integers to use for the chunk sizes on the first axis. These values are *not* validated for correctness, beyond ensuring that the number of items matches the number of partitions. meta : object, optional An optional `meta` parameter can be passed for dask to override the default metadata on the underlying dask array. optimize : bool Whether to optimize the expression before converting to an Array. Returns ------- A Dask Array T)tuplerrrry_validate_chunks_chunksr)rlengthsrQr-optimize_kwargsarrchunkss r to_dask_arrayzFrameBase.to_dask_arrayis]8 d?D//4<<>?Gkk&&sG4  CI rct|jjr>22rcp|jj|}t|dt|dfSNrr)r __divmod__rrrrs rrzFrameBase.__divmod__s3%%e,fQi(.*CCCrcp|jj|}t|dt|dfSr)r __rdivmod__rrs rrzFrameBase.__rdivmod__s3&&u-fQi(.*CCCrc"|jSr)absrs r__abs__zFrameBase.__abs__sxxzrrc |j|}|dk(r$|jtj||||St |j j||||}|j ||SNr)r} numeric_onlyr min_count)rlrrrrr_apply_min_countrrr}rr split_everyrrs rrz FrameBase.sumsy""4( 19&&)# '   flKQU VW$$VY77rc6|r|jj|k\}|j}t|s |j }|j }|j |t j}t|s|jS|S|S)N)r) notnullrrr1 to_serieswherernanmin)rrrcond cond_metas rrzFrameBase._apply_min_counts} <<>%%'94D I!),))+~~'\\$bff\5F!),zz|# Mrc |j|}|dk(r$|jtj||||St |j j||||}|j ||Sr)rlrrprodrrrrs rrzFrameBase.prodsy""4( 19&&)# '   v|[RV WX$$VY77rc `t|d|j|}|jj||||}t |jr4|r2|t |jjdj }t|jj|||||S)Nvar)rr}rT)rr) rkrlrrr0rrrr)rrr}ddofrrrrs rrz FrameBase.vars e,""4( DlK TZZ (\$tzz~~4~@FFGHE JJNN4|N U  rc t|d|j|}|}t|jj ||||}d\} } t |jrY|dk(r|t |j}n|j}|dur,|jjtjg} n |j} | jddgj} t| dkDr|d k(rtt| t|jk7rStt|d t!|tj"gi|j |j$ }nmd} |d k(r)|j'd |j(j*d}| D]} t-|| ||| <n#t/|j} | r t-||}d} | r3| 1| Dcgc]&}t1|j|j(dd(} }|d k(rW| sin"|j(j*|j2d}|j4| st6jnt8f||||d|d|S|j;||||}| r9t |j| |t1|ddt1|dd| d}t8}ni}tj<}|j4|f|d|jd|}|Scc}w)Nstdrr}r rFNrTincludedatetime timedeltar_rr1z datetime64[]unit)rrF)rQrr}r rr)r}r rrr) is_df_like time_colsrrrunitsrQrr)rkrlrsrr r0rrr select_dtypesrnumberrr from_pandasryrzrrastyper'rr+rrrrrr-rsqrt)rrr}r rrr numeric_ddrQneeds_time_conversionrrcolrr_kwargsrsqrt_func_kwargs sqrt_funcs rr z FrameBase.stds e,""4( TZZ(,,f4l- ,7(y TZZ (qy'TZZ(89 '__. t#"((66 {6K"((++Z4M+NVVI9~!19Y3t||3D!D!,4.t4 "?"9$"?"IJ"&**%)$4$4 "J-1)qy%/%6%6TZZ__DUUV7W%X (W*=jov*V 3W%  Y%:IRSAWTZZ]00&$?SES 19-"jjoo C  -:,,28V !&)   <[   !/ ;& w5fd3   7I! I&&&  "     YTs)+K7cl|js tdttj|S)zEnforce the current divisions at runtime. Injects a layer into the Task Graph that checks that the current divisions match the expected divisions at runtime. zNo known divisions to enforce!)rerrrEnforceRuntimeDivisionsrs renforce_runtime_divisionsz#FrameBase.enforce_runtime_divisions`s/ ##=> >d::4@AArc|t|d| td|j|}t|r|jj ||}n|jj }|dk(r#|j tj ||dS|s td|dk7r td |}|jdkDr$|jd d gtjg }tt|d }tt|d}||dzz } | jdk(r| jd} | S)af .. note:: This implementation follows the dask.array.stats implementation of skewness and calculates skewness without taking into account a bias term for finite sample size, which corresponds to the default settings of the scipy.stats skewness calculation. However, Pandas corrects for this, so the values differ by a factor of (n * (n - 1)) ** 0.5 / (n - 2), where n is the number of samples. Further, this method currently does not support filtering out NaN values, which is again a difference to Pandas. skew0`axis=None` isn't currently supported for `skew`rrrF)rQrrbias=False is not implemented. propagate>`nan_policy` other than 'propagate' have not been implemented.rboolrexcluderorderg?)rkrrlr0rr)rrrrrr timedelta64rrXfillna) rrbias nan_policyrrQrm2m3rs rr)zFrameBase.skewjsF* f- <OP P""4( T "&&++L+QD&&++-D 19&&!& ' %&FG G  $%P  ::>''!6*R^^4D(EF52 3 F52 3b#g ;;! ]]3'F rct|d| td|j|}t|r|jj ||}n|jj }|dk(r,t tj |||jdz|dS|s td|dk7r td |}|jdkDr$|jd d gtjg }tt|d }tt|d} | |dzz } | jdk(r| j!d} |r| dz S| S)a .. note:: This implementation follows the dask.array.stats implementation of kurtosis and calculates kurtosis without taking into account a bias term for finite sample size, which corresponds to the default settings of the scipy.stats kurtosis calculation. This differs from pandas. Further, this method currently does not support filtering out NaN values, which is again a difference to Pandas. kurtosisr*r+rF)rQtokenrrr,r-r.rr/r0rr2g@r5r4)rkrrlr0rr=rr _token_prefixrrrrr6rrXr7) rrfisherr8r9rrQrr:m4rs rr=zFrameBase.kurtosissY* j1 <OP P""4( T "&&//T /UD&&//1D 19! ((:5!&  %&FG G  $%P  ::>''!6*R^^4D(EF52 3 F52 3b#g ;;! ]]3'F A: Mrc|j|}t|d|dk(r$|jtj||||S|j j |||}|}|j dk(r|t|j}|j|||}|j|} ttj|| z |d|j } | S) Nsemrr )r}r rr)r}r rrFr) rlrkrrrDrrrrrrrr) rrr}r rrrQrr{rors rrDz FrameBase.sems""4(e, 19&&) ' zz~~V$\~R 99>djj)*E IIV$KI H KKKK 0 GG E"   rc0|d}n|dkr td|jj|||}|r]|jj}t |j t |j k7r|t |j }||fS)Nrzmin_periods must be >= 2)r min_periods)rrcov_get_numeric_datarrr)rrFrrnumericss r_prepare_cov_corrzFrameBase._prepare_cov_corrs  K 1_78 8 LkJ zz335H8##$DLL(99d8#3#345k!!rcZ|j||\}}tt||||Sr)rJrrO)rrFrrscalarrs r_covzFrameBase._covs1"33KN{c%k6JKKrcz|dk7r td|j||\}}tt||||S)Npearsonz-Only Pearson correlation has been implemented)rrJrrN)rrrFrrrLrs r_corrzFrameBase._corrsE Y %&UV V!33KN{d5+{FKLLrc t|d|j|}|dk(r#|jtj|||St |j j ||||S)Nmeanrr}rr)rr)rkrlrrrRrrrrr}rrrs rrRzFrameBase.mean'sq f-""4( 19&&vLt'  IINN6<[tN T  rc |j|}|dk(r#|jtj|||St |j j||||SNrrS)rlrrmaxrrrTs rrWz FrameBase.max5^""4( 19&&fO""4( 19&&quuV$&G GdiimmFK@AArc |j|}|dk(r"|jtj||St |j j||SrZ)rlrrallrrr[s rr^z FrameBase.allEr\rc|j|}|dk(r#|jtj|||St |j j|||SrV)rlrridxminrrrrr}rrs rr`zFrameBase.idxminL^""4( 19&&l' dii..v|[QRRrc|j|}|dk(r#|jtj|||St |j j|||SrV)rlrridxmaxrrras rrdzFrameBase.idxmaxUrbrc |j|}|dk(r#|jtj|||St |j j||||SrV)rlrrrrrrTs rrz FrameBase.min^rXrc|j|}|dk(r"|jtj||St |j j||S)Nr)rr)rlrrrrr)rrrrs rrzFrameBase.countgsP""4( 19&&qww\PT&U UdiioolKHIIrct|dt|jjt |j jS)Nr)rkrsrrrrrs rrz FrameBase.absns8 e,djj!%%'diimmo..rcJt|jj|Sr)rrr)rrs rrzFrameBase.astypeusdii..v677rc|j|dj}t|jj|SNouter)rrr combine_first)rrs rrlzFrameBase.combine_firstys4,,UG<AAdii55e<==rcLt|jj||Sr)rr to_timestamp)rrhows rrnzFrameBase.to_timestamp~sdii44T3?@@rcHt|jjSr)rrisnars rrqzFrameBase.isnasdiinn.//rc 2tjt|ds tdt j ||||}g}t t|D];}|jtt j|||j=|S)a@Pseudorandomly split dataframe into different pieces row-wise Parameters ---------- frac : list List of floats that should sum to one. random_state : int or np.random.RandomState If int or None create a new RandomState with this as the seed. Otherwise draw from the passed RandomState. shuffle : bool, default False If set to True, the dataframe is shuffled (within partition) before the split. Examples -------- 50/50 split >>> a, b = df.random_split([0.5, 0.5]) # doctest: +SKIP 80/10/10 split, consistent random_state >>> a, b, c = df.random_split([0.8, 0.1, 0.1], random_state=123) # doctest: +SKIP See Also -------- dask.DataFrame.sample rzfrac should sum to 1) rallcloserrrSplitrrappendr SplitTaker)rfrac random_staterrr/rs r random_splitzFrameBase.random_splits|:{{3t9a(34 4 4|W=s4y! LA JJ~dnnUAtyy&IJ K L rcJt|jj|Sr)rrround)rdecimalss rr{zFrameBase.roundsdiiooh788rc|j|}|j|}t|tr |jn|}t|tr |jn|}t |jj ||Sr)rrrrrrrrrs rrzFrameBase.wheresd++D1,,U3&tY7tyyT( : diioodE:;;rc|j|}|j|}t|tr |jn|}t|tr |jn|}t |jj ||Sr)rrrrrmaskr~s rrzFrameBase.masksd++D1,,U3&tY7tyyT( : diinnT59::rcNt|jj|||Sr)rrr)r to_replaceriregexs rrzFrameBase.replaces dii// E5IJJrct|}|dk(r"|jtj||S|}|t |dd}t t ||S)NrrlimitffillcyNrrrs rrz!FrameBase.ffill..sr)rlrrrrBrrArrrrs rrzFrameBase.ffillsYd# 19&&qwwT&G G =g{;EeE5122rct|}|dk(r"|jtj||S|}|t |dd}t t ||S)Nrrbfillc |jdz SNrr1rs rrz!FrameBase.bfill..s9Jr)rlrrrrBrr>rs rrzFrameBase.bfillsZd# 19&&qwwT&G G =g/JKEeE5122rc|j|}|dk(r"|jtj||St |t r |j }t|j j|SNrr)rlrrr7rrrr)rrirs rr7zFrameBase.fillnasb""4( 19&&qxxT&B B eY 'JJEdii..u566rct|ts tdt|}|dk(rt t |||S|j t jdd|||S)Nzperiods must be an integerrF)rrrperiodsrr)rr rrlrrDrr)rrrrs rshiftzFrameBase.shiftsk'8,89 9d# 19!%gt"<= =""" % #  rct|}|dk(rtt||S|jtjddd||S)a| .. note:: Pandas currently uses an ``object``-dtype column to represent boolean data with missing values. This can cause issues for boolean-specific operations, like ``|``. To enable boolean- specific operations, at the cost of metadata that doesn't match pandas, use ``.astype(bool)`` after the ``shift``. rF)rrrrrr)rlrr?rr)rrrs rdiffzFrameBase.diffsUd# 19!$tW"56 6""" %! #  rcPt|jj||||Sr)rr rename_axis)rmapperrrrs rrzFrameBase.rename_axis s$dii33FE7DQRRrctt|s,t|s t|r|dvrd}nd}t||}|S)N)innerrrrr1)r#r1r0r)rrjoinrs rrz!FrameBase._create_alignable_frames@!%( 5 !%6u%=((  ;?E rcv|j||}|jj|j|||Sr)rralign)rrrrrs rrzFrameBase.aligns1,,UD9yyuzz4zBBrcLt|jj|S)a9Approximate number of unique rows. This method uses the HyperLogLog algorithm for cardinality estimation to compute the approximate number of unique rows. The approximate error is 0.406%. Parameters ---------- split_every : int, optional Group partitions into groups of this size while performing a tree-reduction. If set to False, no tree-reduction will be used. Default is 8. Returns ------- a float representing the approximate number of elements r)rrnunique_approxrrs rrzFrameBase.nunique_approx"s!$dii66;6OPPrc |dk(r"|jtj||St|jj|SNr)rr}r|)rrcumsumrrrrr}rs rrzFrameBase.cumsum6B 19&&qxxd6&J Jdii..f.=>>rc |dk(r"|jtj||St|jj|Sr)rrcumprodrrrs rrzFrameBase.cumprod<sB 19&&qyytF&K Kdii//v/>??rc|dk(r"|jtj||St|jj|Sr)rrcummaxrrrrr}s rrzFrameBase.cummaxBrrc|dk(r"|jtj||St|jj|Sr)rrcumminrrrs rrzFrameBase.cumminHrrc ||dkr|dur td| | r td|r|jni}|j| ||d<| r| jni} | j| |xs|xs|| d<|r|jni}|j| |xs||d<tt ||||| ||S)aPGeneric row-wise reductions. Parameters ---------- chunk : callable Function to operate on each partition. Should return a ``pandas.DataFrame``, ``pandas.Series``, or a scalar. aggregate : callable, optional Function to operate on the concatenated result of ``chunk``. If not specified, defaults to ``chunk``. Used to do the final aggregation in a tree reduction. The input to ``aggregate`` depends on the output of ``chunk``. If the output of ``chunk`` is a: - scalar: Input is a Series, with one row per partition. - Series: Input is a DataFrame, with one row per partition. Columns are the rows in the output series. - DataFrame: Input is a DataFrame, with one row per partition. Columns are the columns in the output dataframes. Should return a ``pandas.DataFrame``, ``pandas.Series``, or a scalar. combine : callable, optional Function to operate on intermediate concatenated results of ``chunk`` in a tree-reduction. If not provided, defaults to ``aggregate``. The input/output requirements should match that of ``aggregate`` described above. $META token : str, optional The name to use for the output keys. split_every : int, optional Group partitions into groups of this size while performing a tree-reduction. If set to False, no tree-reduction will be used, and all intermediates will be concatenated and passed to ``aggregate``. Default is 8. chunk_kwargs : dict, optional Keyword arguments to pass on to ``chunk`` only. aggregate_kwargs : dict, optional Keyword arguments to pass on to ``aggregate`` only. combine_kwargs : dict, optional Keyword arguments to pass on to ``combine`` only. kwargs : All remaining keywords will be passed to ``chunk``, ``combine``, and ``aggregate``. Examples -------- >>> import pandas as pd >>> import dask.dataframe as dd >>> df = pd.DataFrame({'x': range(50), 'y': range(50, 100)}) >>> ddf = dd.from_pandas(df, npartitions=4) Count the number of rows in a DataFrame. To do this, count the number of rows in each partition, then sum the results: >>> res = ddf.reduction(lambda x: x.count(), ... aggregate=lambda x: x.sum()) >>> res.compute() x 50 y 50 dtype: int64 Count the number of rows in a Series with elements greater than or equal to a value (provided via a keyword). >>> def count_greater(x, value=0): ... return (x >= value).sum() >>> res = ddf.x.reduction(count_greater, aggregate=lambda x: x.sum(), ... chunk_kwargs={'value': 25}) >>> res.compute() np.int64(25) Aggregate both the sum and count of a Series at the same time: >>> def sum_and_count(x): ... return pd.Series({'count': x.count(), 'sum': x.sum()}, ... index=['count', 'sum']) >>> res = ddf.x.reduction(sum_and_count, aggregate=lambda x: x.sum()) >>> res.compute() count 50 sum 1225 dtype: int64 Doing the same, but for a DataFrame. Here ``chunk`` returns a DataFrame, meaning the input to ``aggregate`` is a DataFrame with an index with non-unique entries for both 'x' and 'y'. We groupby the index, and sum each group to get the final result. >>> def sum_and_count(x): ... return pd.DataFrame({'count': x.count(), 'sum': x.sum()}, ... columns=['count', 'sum']) >>> res = ddf.reduction(sum_and_count, ... aggregate=lambda x: x.groupby(level=0).sum()) >>> res.compute() count sum x 50 1225 y 50 3725 rFzsplit_every must be at least 2z+`combine_kwargs` provided with no `combine`r)rrupdaterrP) rchunk aggregatecombinerQr>r chunk_kwargsaggregate_kwargscombine_kwargsrs r reductionzFrameBase.reductionNs`  "{Q;e;S=> > ? !NOO.:|((* F#$ V2@,,.bf%!(!>I!>v6F+002B'#,#5      rcLt|jj||S)atReturn the memory usage of each partition Parameters ---------- index : bool, default True Specifies whether to include the memory usage of the index in returned Series. deep : bool, default False If True, introspect the data deeply by interrogating ``object`` dtypes for system-level memory consumption, and include it in the returned values. Returns ------- Series A Series whose index is the partition number and whose values are the memory usage of each partition in bytes. )rrmemory_usage_per_partition)rrrws rrz$FrameBase.memory_usage_per_partitions &diiBB5$OPPrcddlm}||S)zPurely label-location based indexer for selection by label. >>> df.loc["b"] # doctest: +SKIP >>> df.loc["b":"d"] # doctest: +SKIP r) LocIndexer)"dask.dataframe.dask_expr._indexingr)rrs rrz FrameBase.locs B$rc>ttj|Sr)rrNotNullrs rrzFrameBase.notnullsdll4011rc$|jSr)rrs risnullzFrameBase.isnull s rc ||jr|r|S|jS| |r td|||}n |j}t ||\}t dfz}|s|St t |jdz k(rGtfdtdt Dsttj||Sttj||S)aeCompute the current divisions of the DataFrame. This method triggers immediate computation. If you find yourself running this command repeatedly for the same dataframe, we recommend storing the result so you don't have to rerun it. If the column or index values overlap between partitions, raises ``ValueError``. To prevent this, make sure the data are sorted by the column or index. Parameters ---------- col : string, optional Calculate the divisions for a non-index column by passing in the name of the column. If col is not specified, the index will be used to calculate divisions. In this case, if the divisions are already known, they will be returned immediately without computing. set_divisions : bool, default False Whether to set the computed divisions into the DataFrame. If False, the divisions of the DataFrame are unchanged. Examples -------- >>> import dask >>> ddf = dask.datasets.timeseries(start="2021-01-01", end="2021-01-07", freq="1h").clear_divisions() >>> divisions = ddf.compute_current_divisions() >>> print(divisions) # doctest: +NORMALIZE_WHITESPACE (Timestamp('2021-01-01 00:00:00'), Timestamp('2021-01-02 00:00:00'), Timestamp('2021-01-03 00:00:00'), Timestamp('2021-01-04 00:00:00'), Timestamp('2021-01-05 00:00:00'), Timestamp('2021-01-06 00:00:00'), Timestamp('2021-01-06 23:00:00')) >>> ddf = ddf.reset_index().clear_divisions() >>> divisions = ddf.compute_current_divisions("timestamp") >>> print(divisions) # doctest: +NORMALIZE_WHITESPACE (Timestamp('2021-01-01 00:00:00'), Timestamp('2021-01-02 00:00:00'), Timestamp('2021-01-03 00:00:00'), Timestamp('2021-01-04 00:00:00'), Timestamp('2021-01-05 00:00:00'), Timestamp('2021-01-06 00:00:00'), Timestamp('2021-01-06 23:00:00')) >>> ddf = ddf.set_index("timestamp", divisions=divisions, sorted=True) z9Can't set divisions of non-index, call set_index instead.) allow_overlaprrc3:K|]}||dz k\ywrNr)rrmaxesminss rrz6FrameBase.compute_current_divisions..I s"L1tAw%A,.L) rerrr_compute_partition_statsrrrrrr SetDivisionsResolveOverlappingDivisions)rr! set_divisionsrlensrrrs @@rcompute_current_divisionsz#FrameBase.compute_current_divisions s` ;4// >> ! ?}%K  ?IEJJE4U-XeT$K59,.   t9DNN+a/ /LaT8KLL%d&7&7i&HIId>>tT5RVWXXrr)rorientrrc"t|||||S)z Construct a Dask DataFrame from a Python Dictionary See Also -------- dask.dataframe.from_dict )rr) from_dict)clsrrrrrs rrzFrameBase.from_dictN s{F%QQrc(ddlm}|||g|i|S)z-See dd.to_json docstring for more informationr)to_json)dask.dataframe.ior)rr2rrrs rrzFrameBase.to_json[ s-tX7777rc 8ddlm} | |||||||||| | | |  S)Nr)to_sql) schema if_existsr index_label chunksizerrrparallel engine_kwargs)dask.dataframe.io.sqlr)rrurirrrrrrrrrrrs rrzFrameBase.to_sqla s; 1   #'  rc(ddlm}|||g|i|S)z,See dd.to_orc docstring for more informationr)to_orc)dask.dataframe.io.orcr)rpathrrrs rrzFrameBase.to_orc s0dD242622rc "ddlm}|||fi|S)z,See dd.to_csv docstring for more informationr)to_csv)dask.dataframe.io.csvr)rr2rrs rrzFrameBase.to_csv s0dH///rcddlm}|dur,t|jtj }||}|j ||}|df|_|S)Nr) to_recordsT)#dask.dataframe.dask_expr.io.recordsrrrrrrr)rrrrrecordsrs rrzFrameBase.to_records sXB d?D//4<<>?GT"&&w8!!9,rcnddlm}ddlm}t ||rt |}t ||jk7r$tdt |d|j|jdk(r ||f}|S||t |jff}|S|td|d|jS) Nr)Sequence)normalize_chunkszJThe number of items in 'lengths' does not match the number of partitions.  != rz!Unexpected value for 'lengths': '') collections.abcrdask.array.corerrrrrrrrr)rrrrrrs rrzFrameBase._validate_chunks s,4 gx (GnG7|t/// ##&w<.T5E5E4FH yyA~)7*5M*7S5F4H*IJM  @ KL L{{rc$ddlm}||||S)zCreate a Dask Bag from a Seriesr)to_bag)r)dask.dataframe.dask_expr.io.bagr)rrrrs rrzFrameBase.to_bag s:dE&11rc (ddlm}||||||fi|S)z,See dd.to_hdf docstring for more informationr)to_hdf)dask.dataframe.io.hdfr)r path_or_bufr\moderurrs rrzFrameBase.to_hdf s0dKdFEfEErcddlm}|r|j}n|}|j}|j }d|j z}|j ||d}|Dcgc]}t|||c}Scc}w)aConvert into a list of ``dask.delayed`` objects, one per partition. Parameters ---------- optimize_graph : bool, optional If True [default], the graph is optimized before converting into ``dask.delayed`` objects. Examples -------- >>> partitions = df.to_delayed() # doctest: +SKIP See Also -------- dask_expr.from_delayed r)HighLevelGraphrr) dependencies)layer)dask.highlevelgraphrr-rBr<r from_collectionsr|)roptimize_graphrrrgraphrks r to_delayedzFrameBase.to_delayed sx" 7 MMOEE""$$$&U[[(//u2/N8<=15.===s#A;c xddlm}|xs |j}|j|}|j|fi|S)a1Move to a new DataFrame backend Parameters ---------- backend : str, Optional The name of the new backend to move to. The default is the current "dataframe.backend" configuration. Returns ------- DataFrame, Series or Index rr5)"dask.dataframe.dask_expr._backendsr6backenddispatch to_backend)rrrr6backend_entrypoints rrzFrameBase.to_backend sH S@8@@A8AA'J,!,,T. s!%%u%-rFr|) rrrrrrdotgroupbyrapplyr _dot_seriesr)rrrQss rr z FrameBase.dot s%+IJ J eY '##AEE5t#TJIr)returnz expr.ExprrTNN)r2 str | Nonerr rNone)fusedN)r8rrr )r,r/F)rcr/)rlr/)rrT)rorrr/)rT)rwr/) rzstr | list | no_defaultrr/r int | Nonerr rr/rr/)NNNNF)rz tuple | Nonerr$rr rr/)NNT)r-r/rr!)rTFrF)rTrFF)rTr-F)rTTr-F)NTrFF)NFFF)rONFFF)rTFFrTF)rFF)NrNFr)rN)rNr)rr)rk)rkNNrTTF)rr/rwr/)rr/) NfailTNNNNTFN)rrrrrrrr/r )Fr)aF)rr )NNFN)r __module__ __qualname____doc__ staticmethodr$get__dask_scheduler____dask_optimize__rpropertyrrrZcached_propertyrrrrrrsetterrrrrrr r#r%r(r.r5r7r:r=r<rBrDrFr-rKrRrTr_rbrerrmrqrtrrrrrrrrrIrrrrxrrr4rryrrrrrrproductrr r'r)r=kurtrDrJrMrPrRrWrr^r`rdrrrrrlrnrqryr{rrrrrrrr7rrrrrrrrrrrrrrrr classmethodrrrrrrrrrrrrr rrrr __classcell__r s@rrr*s 0%Y(8(@A%%DE))##.%%&&.."" ^^77 93LL-<65 & (7B6"/H "%%$ # 48=(     -))// \\;,()&",,4  4 l <//*"2'1""&%)n #nn n # n  nn`",,A A /<#r*  N +N `#r* Z +Z |#'"&%)  __ _# _  _D9='15' 'R 3 3DD",, 8 8, ",, 8 8*G",,     &",,  ] ]~B",,  8 8t",,  = =~ D",,NS : " OTL M",,CH     ",,V V",,B B ",,B B ",,S S",,S S",,V V",,J J ",,/ / ",,8 8",,> >",,A A",,0 0$L",,9 9",, "< <",,!vv; ;",,!%ZuK K",,3 3",,3 3",,7 7",,   "",,   ,",,z:AS S ",,C CQ(",,? ? ",,@ @ ",,? ? ",,? ? M ^Q*  ",,2 2",, GYR"#IT4 R R8       B3 0 22 F >:=,"))( T T",, /  /. `"rrcPt|dtj|i|Sr)rzrr )rrs rrrJ s) , "47 +AEE4,B6,B CCr)__add____radd____sub____rsub____mul____rmul____mod____rmod__ __floordiv__ __rfloordiv__ __truediv__ __rtruediv____pow____rpow____lt____rlt____gt____rgt____le____rle____ge____rge____eq____ne____and____rand____or____ror____xor____rxor__)r) __invert____neg____pos__c|eZdZUdZeZded<ejZ e dZ e dZ e dZ eejdZe dZd Zd Zd Zeejd ZeejdSd Ze dZdZdTdZdZdZeejdZeejdUdZ dVdZeej dWdZeej dXdZdZ dZ!dZ"fdZ#dZ$dYdZ%eejdZdZ&eejdZd Z'eejd[d!Z(eejd\d"Z)eejd#$ d]d%Z*e+d&'e,d(d)d*Z-eeje,de,fd+Z.e/d^d_d,Z0eejd-g.dYd/Z1eejdTd0Z2eejd1Z3eejd`d2Z4d3Z5eejdYd4Z6eejd5Z7 da dbd6Z8 dc ddd7Z9d8Z:eejded9Z;eejd:Z<eejd;Z=dfd<Z>e d=Z?d>Z@dgd?ZAdgd@ZBdgdAZCdgdBZDdgdCZEdgdDZFdUdEZGeejdhdFZHdidGZIeejdjdHZJdkdIZKeej dldJZLeejdKZMdmdLZNeejdmdMZOeej dndNZPeejdodOZQeejdodPZReej dpdQZSdRZTxZUS)qrzDataFrame-like Expr Collection. The constructor takes the expression that represents the query as input. The class is not meant to be instantiated directly. Instead, use one of the IO connectors from Dask. ClassVar[set[str]] _accessorsc|jtt|jdzt|jfSr)rrWrrrs rshapezDataFrame.shape s/yyCDLL 1155s4<<7HHHrcy)Return dimensionalityrrrs rrzDataFrame.ndim rctd)NzChecking whether a Dask DataFrame has any rows may be expensive. However, checking the number of columns is fast. Depending on which of these results you need, use either `len(df.index) == 0` or `len(df.columns) == 0`)rurs remptyzDataFrame.empty s+ =  rc#tKt|jD]\}}||jdd|ffywr) enumeraterr)rrrs ritemszDataFrame.items s9 . *GAt1a4) ) *s68c2|j|jgSrrrs raxeszDataFrame.axes s DLL))rc||jvSr)r)rr\s r __contains__zDataFrame.__contains__ sdjj  rc,t|jSr)iterrrs r__iter__zDataFrame.__iter__ sDJJrcddlm}||S)Nr)DaskDataFrameInterchange)%dask.dataframe.dask_expr._interchangerr)rrrrrs r __dataframe__zDataFrame.__dataframe__ sR'--rc#K|j}t|jD]9}|j|j }|j Ed{;y7wr)r-rrrriterrows)rrrdfs rrvzDataFrame.iterrows sV t''( %A$$Q'//1B{{} $ $ % $sAA'A%A'c#K|j}t|jD]<}|j|j }|j ||Ed{>y7wNrr)r-rrrr itertuples)rrrrrrws rr{zDataFrame.itertuples s] t''( =A$$Q'//1B}}5t}< < < = $>?F rc |j|}|dk(r#|jtj|||St |j j|||Sr)rlrrcliprrrlowerupperrrs rrzDataFrame.clip sS""4( 19&&qvvue$&G GdiinnUE4@AArc 2t|||||||||| | | |  S)u7Merge the DataFrame with another DataFrame This will merge the two datasets, either on the indices, a certain column in each dataset or the index in one dataset and the column in another. Parameters ---------- right: dask.dataframe.DataFrame how : {'left', 'right', 'outer', 'inner', 'leftsemi'}, default: 'inner' How to handle the operation of the two objects: - left: use calling frame's index (or column if on is specified) - right: use other frame's index - outer: form union of calling frame's index (or column if on is specified) with other frame's index, and sort it lexicographically - inner: form intersection of calling frame's index (or column if on is specified) with other frame's index, preserving the order of the calling's one - leftsemi: Choose all rows in left where the join keys can be found in right. Won't duplicate rows if the keys are duplicated in right. Drops all columns from right. on : label or list Column or index level names to join on. These must be found in both DataFrames. If on is None and not merging on indexes then this defaults to the intersection of the columns in both DataFrames. left_on : label or list, or array-like Column to join on in the left DataFrame. Other than in pandas arrays and lists are only support if their length is 1. right_on : label or list, or array-like Column to join on in the right DataFrame. Other than in pandas arrays and lists are only support if their length is 1. left_index : boolean, default False Use the index from the left DataFrame as the join key. right_index : boolean, default False Use the index from the right DataFrame as the join key. suffixes : 2-length sequence (tuple, list, ...) Suffix to apply to overlapping column names in the left and right side, respectively indicator : boolean or string, default False If True, adds a column to output DataFrame called "_merge" with information on the source of each row. If string, column with information on source of each row will be added to output DataFrame, and column will be named value of string. Information column is Categorical-type and takes on a value of "left_only" for observations whose merge key only appears in `left` DataFrame, "right_only" for observations whose merge key only appears in `right` DataFrame, and "both" if the observation’s merge key is found in both. npartitions: int or None, optional The ideal number of output partitions. This is only utilised when performing a hash_join (merging on columns only). If ``None`` then ``npartitions = max(lhs.npartitions, rhs.npartitions)``. Default is ``None``. shuffle_method: {'disk', 'tasks', 'p2p'}, optional Either ``'disk'`` for single-node operation or ``'tasks'`` and ``'p2p'``` for distributed operation. Will be inferred by your current scheduler. broadcast: boolean or float, optional Whether to use a broadcast-based join in lieu of a shuffle-based join for supported cases. By default, a simple heuristic will be used to select the underlying algorithm. If a floating-point value is specified, that number will be used as the ``broadcast_bias`` within the simple heuristic (a large number makes Dask more likely to choose the ``broacast_join`` code path). See ``broadcast_join`` for more information. Notes ----- There are three ways to join dataframes: 1. Joining on indices. In this case the divisions are aligned using the function ``dask.dataframe.multi.align_partitions``. Afterwards, each partition is merged with the pandas merge function. 2. Joining one on index and one on column. In this case the divisions of dataframe merged by index (:math:`d_i`) are used to divide the column merged dataframe (:math:`d_c`) one using ``dask.dataframe.multi.rearrange_by_divisions``. In this case the merged dataframe (:math:`d_m`) has the exact same divisions as (:math:`d_i`). This can lead to issues if you merge multiple rows from (:math:`d_c`) to one row in (:math:`d_i`). 3. Joining both on columns. In this case a hash join is performed using ``dask.dataframe.multi.hash_join``. In some cases, you may see a ``MemoryError`` if the ``merge`` operation requires an internal ``shuffle``, because shuffling places all rows that have the same index in the same partition. To avoid this error, make sure all rows with the same ``on``-column value can fit on a single partition. )r broadcast)merge) rrrorleft_onright_on left_index right_indexsuffixes indicatorrrrs rrzDataFrame.merge% s9V           #  rc t|tst|s t|d}t|tsIt |j s4t |j drttj|}t|tst|trtd|Ds td|dvr td|Dcgc]}t|s t|dn|}}tt|jg|Dcgc]}|jc}z|S|j||dud ||||f|| Scc}wcc}w) Nrr1rc3<K|]}t|tywrrr)ros rrz!DataFrame.join.. s6-. 1i(6rz-other must be DataFrame or list of DataFrames)rkrz-merge_multi only supports left or outer joins)roT)rrrrrorrr)rrr#rr0rhasattrrrToFramerr^rrJr) rrrrolsuffixrsuffixrrrs rrzDataFrame.join sO%&/A%/H15E5$'%ekk2 V,"4<<#67E%+eT*#62763!!PQQ++ !PQQ6H5J A1-PQQE "tyykU,CQVV,CCM zzTzw')#   -Ds !EEc ddlm}t|trt|tst d|d|||f||||d|S)Nr)GroupByz3`by` must be a column name or list of columns, got .) group_keyssortobserveddropna)!dask.dataframe.dask_expr._groupbyrrrrr)rr rrrrrrs rr zDataFrame.groupby se > b) $ZF-CEbTK    "    rc t|ttfrNt|tr>|jdit ||j Dcic] \}}||| c}}}nt|tjr@t|ts0|jditjt||}n|t|s!t|st|ttfr|j||}nc*  DI   tS% 0 G s A AAc tj||S#t$rj} ||jjj vrt |j|cYd}~S|#t$rt|!|cYcYd}~SwxYwd}~wwxYwr) rVrWrXrrrrrar_)rr\r]r s rr_zDataFrame.__getattr__ s~ 0**45 5 0 0$))//111)$))C.99 ! 0w*3// 0 0s> B 9A$B "A$$B<B=B BBB c.ttt|}|j|j|jttt j |jd|jDt|S)Nc3dK|](}t|ts|js%|*ywr)rr isidentifierrs rrz$DataFrame.__dir__..! s"VqZ3-?ANNDTVs 000) rdirr r__dict__rrrrrrs r__dir__zDataFrame.__dir__ s\ DJ   S^$% VDLLVVAwrc tstdtd|Htjt j |||d}tjt|dttj||||S)Nz1DataFrame.map requires pandas>=2.1.0, but pandas=z is installed.T na_actionudfmaprrrrQ) r&rr(remulaterrrrr2rMap)rrrrQs rrz DataFrame.map$ st%CNCST  <<<tTYDQD MM,tE: ;dhhtQUVWWrc2tt||||SN)ro_columnsrrrYrrorrs rnlargestzDataFrame.nlargest/ s TQk J  rc2tt||||SrrrZrs r nsmallestzDataFrame.nsmallest5 s da'{ K  rc0tt|||SN)rw_indexrrVrrwrs r memory_usagezDataFrame.memory_usage; .t$uMNNrc |j|d}tj|j|js"ttj|||||Sttj |||||Srj)rrrrCombineFrameAlign CombineFrame)rrrr overwrites rrzDataFrame.combine? st,,UG<""499ejj9!&&tUD*iP    dE4Y G  rz/keep=False will raise a ``NotImplementedError``)inconsistenciesc t|}|dur tdt|}t|jj ||t t|||||||S)NFdrop_duplicates with keep=False)subsetkeep)rr split_outrrr)rhrrgrsrdrop_duplicatesrrQ)rrrrrrrs rrzDataFrame.drop_duplicatesJ sq7~F 5=%&GH H!&)djj!11d1K )#'-   rrrrrQrc@|j|}|dk(r d}t||turHtjt j ||f|d|d|}tjt|t|jj |g|||d|S)aParallel version of pandas.DataFrame.apply This mimics the pandas version except for the following: 1. Only ``axis=1`` is supported (and must be specified explicitly). 2. The user should provide output metadata via the `meta` keyword. Parameters ---------- func : function Function to apply to each column/row axis : {0 or 'index', 1 or 'columns'}, default 0 - 0 or 'index': apply function to each column (NOT SUPPORTED) - 1 or 'columns': apply function to each row $META args : tuple Positional arguments to pass to function in addition to the array/series Additional keyword arguments will be passed as keywords to the function Returns ------- applied : Series or DataFrame Examples -------- >>> import pandas as pd >>> import dask.dataframe as dd >>> df = pd.DataFrame({'x': [1, 2, 3, 4, 5], ... 'y': [1., 2., 3., 4., 5.]}) >>> ddf = dd.from_pandas(df, npartitions=2) Apply a function to row-wise passing in extra arguments in ``args`` and ``kwargs``: >>> def myadd(row, a, b=1): ... return row.sum() + a + b >>> res = ddf.apply(myadd, axis=1, args=(2,), b=1.5) # doctest: +SKIP By default, dask tries to infer the output metadata by running your provided function on some fake data. This works well in many cases, but can sometimes be expensive, or even fail. To avoid this, you can manually specify the output metadata with the ``meta`` keyword. This can be specified in many forms, for more information see ``dask.dataframe.utils.make_meta``. Here we specify the output is a Series with name ``'x'``, and dtype ``float64``: >>> res = ddf.apply(myadd, axis=1, args=(2,), b=1.5, meta=('x', 'f8')) In the case where the metadata doesn't change, you can also pass in the object itself directly: >>> res = ddf.apply(lambda row: row + 1, axis=1, meta=ddf) See Also -------- DataFrame.map_partitions rzGDask DataFrame.apply only supports axis=1 Try: df.apply(func, axis=1)T)rrrr) rlrrIrrrr rrr2r)rfunctionrQrrrrs rr zDataFrame.applyi s|""4( 190 &c* * : <<x.24KQD MM,t, - DIIOOH Lt L$T LV L  rc|tur|tur tdt|}tt j ||||S)NzBYou cannot set both the how and thresh arguments at the same time.)rorthresh)rIrrgrr DropnaFrame)rrorrs rrzDataFrame.dropna sN j V:%=T "&)   Ts6& I  rc\|dvrtd||rddd}|j||S|S)N)rrrrNNo axis named rrrrr0rr numeric_axisnum_axiss rrlzDataFrame._validate_axis sA 7 7~dV45 5 BCPQ8RH<<d+ +Krrua_argsc\| tdttj||S)NzCannot rename index.r)rrr RenameFrame)rrrs rrenamezDataFrame.rename s,  34 4d..tWEFFrc|dvr,t|jdk(r||jdS|S|dk(rtt|dt d|dt|)Nrrrz& does not support squeeze along axis 0zNo axis z for object type )rrrr r)rrs rsqueezezDataFrame.squeeze su 9 4<< A%DLLO,, QY%:,DE  xv->tDzlKL LrcXt|}ttj||S)N)column)rgrr ExplodeFrame)rr s rexplodezDataFrame.explode s$!&)d//VDEErc| | tdt|}|dk(r|xs|}n|dk(r | tdtt j |||S)Nz)must either specify 'columns' or 'labels'rrz@Drop currently only works for axis=1 or when columns is not None)rerrors)rrlrrrDrop)rlabelsrrrs rrlzDataFrame.drop sg ?v~GH Hd# 19'G QY7?%R diigfMNNrc "ddlm}|||fi|S)Nr) to_parquet)#dask.dataframe.dask_expr.io.parquetr)rrrrs rrzDataFrame.to_parquet sB$///rct|jj||j}t |j |S)Nr0)rrrrrr)rrr1rs rrzDataFrame.select_dtypes s? JJ $ $Wg $ F N N dii011rc Nd|vr tdtt|||S)Ninplacez!inplace is not supported for eval)r4 expr_kwargs)rrr@rrrs revalzDataFrame.eval s*  %&IJ Jd4tHIIrc  2t|trt|dk(r|d}t|tr,td|Dr t dt d|dt|t r"t dt|jdt|tr%|j|jjk(r|S||jjk(r|S| t||s|s | td |rU|(t|dz |jk7r d } t| tt!||||| } | j#d S|stt!|||d| Stt%||||||| t'|| |  S)aBSet the DataFrame index (row labels) using an existing column. If ``sort=False``, this function operates exactly like ``pandas.set_index`` and sets the index on the DataFrame. If ``sort=True`` (default), this function also sorts the DataFrame by the new index. This can have a significant impact on performance, because joins, groupbys, lookups, etc. are all much faster on that column. However, this performance increase comes with a cost, sorting a parallel dataset requires expensive shuffles. Often we ``set_index`` once directly after data ingest and filtering and then perform many cheap computations off of the sorted dataset. With ``sort=True``, this function is much more expensive. Under normal operation this function does an initial pass over the index column to compute approximate quantiles to serve as future divisions. It then passes over the data a second time, splitting up each input partition into several pieces and sharing those pieces to all of the output partitions now in sorted order. In some cases we can alleviate those costs, for example if your dataset is sorted already then we can avoid making many small pieces or if you know good values to split the new index column then we can avoid the initial pass over the data. For example if your new index is a datetime index and your data is already sorted by day then this entire operation can be done for free. You can control these options with the following parameters. Parameters ---------- other: string or Dask Series Column to use as index. drop: boolean, default True Delete column to be used as the new index. sorted: bool, optional If the index column is already sorted in increasing order. Defaults to False npartitions: int, None, or 'auto' The ideal number of output partitions. If None, use the same as the input. If 'auto' then decide by memory use. Only used when ``divisions`` is not given. If ``divisions`` is given, the number of output partitions will be ``len(divisions) - 1``. divisions: list, optional The "dividing lines" used to split the new index into partitions. For ``divisions=[0, 10, 50, 100]``, there would be three output partitions, where the new index contained [0, 10), [10, 50), and [50, 100), respectively. See https://docs.dask.org/en/latest/dataframe-design.html#partitions. If not given (default), good divisions are calculated by immediately computing the data and looking at the distribution of its values. For large datasets, this can be expensive. Note that if ``sorted=True``, specified divisions are assumed to match the existing partitions in the data; if this is untrue you should leave divisions empty and call ``repartition`` after ``set_index``. inplace: bool, optional Modifying the DataFrame in place is not supported by Dask. Defaults to False. sort: bool, optional If ``True``, sort the DataFrame by the new index. Otherwise set the index on the individual existing partitions. Defaults to ``True``. shuffle_method: {'disk', 'tasks', 'p2p'}, optional Either ``'disk'`` for single-node operation or ``'tasks'`` and ``'p2p'`` for distributed operation. Will be inferred by your current scheduler. compute: bool, default False Whether or not to trigger an immediate computation. Defaults to False. Note, that even if you set ``compute=False``, an immediate computation will still be triggered if ``divisions`` is ``None``. partition_size: int, optional Desired size of each partitions in bytes. Only used when ``npartitions='auto'`` Examples -------- >>> import dask >>> ddf = dask.datasets.timeseries(start="2021-01-01", end="2021-01-07", freq="1h").reset_index() >>> ddf2 = ddf.set_index("x") >>> ddf2 = ddf.set_index(ddf.x) >>> ddf2 = ddf.set_index(ddf.timestamp, sorted=True) A common case is when we have a datetime column that we know to be sorted and is cleanly divided by day. We can set this index for free by specifying both that the column is pre-sorted and the particular divisions along which is is separated >>> import pandas as pd >>> divisions = pd.date_range(start="2021-01-01", end="2021-01-07", freq='1D') >>> divisions DatetimeIndex(['2021-01-01', '2021-01-02', '2021-01-03', '2021-01-04', '2021-01-05', '2021-01-06', '2021-01-07'], dtype='datetime64[ns]', freq='D') Note that ``len(divisions)`` is equal to ``npartitions + 1``. This is because ``divisions`` represents the upper and lower bounds of each partition. The first item is the lower bound of the first partition, the second item is the lower bound of the second partition and the upper bound of the first partition, and so on. The second-to-last item is the lower bound of the last partition, and the last (extra) item is the upper bound of the last partition. >>> ddf2 = ddf.set_index("timestamp", sorted=True, divisions=divisions.tolist()) If you'll be running `set_index` on the same (or similar) datasets repeatedly, you could save time by letting Dask calculate good divisions once, then copy-pasting them to reuse. This is especially helpful running in a Jupyter notebook: >>> ddf2 = ddf.set_index("name") # slow, calculates data distribution >>> ddf2.divisions # doctest: +SKIP ["Alice", "Laura", "Ursula", "Zelda"] >>> # ^ Now copy-paste this and edit the line above to: >>> # ddf2 = ddf.set_index("name", divisions=["Alice", "Laura", "Ursula", "Zelda"]) rrc3<K|]}t|tywrrrs rrz&DataFrame.set_index.. s;:a+;rz*List[FrameBase] not supported by set_indexzWDask dataframe does not yet support multi-indexes. You tried to index with this index: z% Indexes must be single columns only.zgDask dataframe does not yet support multi-indexes. You tried to index with a frame with these columns: NzfSpecifying npartitions with sort=False or sorted=True is not supported. Call `repartition` afterwards.aeWhen doing `df.set_index(col, sorted=True, divisions=...)`, divisions indicates known splits in the index column. In this case divisions must be the same length as the existing divisions in `df` If the intent is to repartition into new divisions after setting the index, you probably want: `df.set_index(col, sorted=True).repartition(divisions=divisions)`) new_divisionsruT)r)ru)user_divisionsrupsamplerrrur)rrrrrrrrrr rrr.rrrrbrrarm)rrrlrOrrrrr rrurrrs r set_indexzDataFrame.set_index st eT "s5zQ!HE eT ";U;; LMM);;@'B;; eY '%GGKEMMGZF[\77  eV ${{djj... djjoo %K  I & $K$;<  $Y!);t?O?O)OX!o%#!%YvF 33$3G G!!$tT&I  ('!-4^D   rc  Z|dvr tdt|ts|g}td|Drt dt |zt|t s(t|t|k(std|d|tt|||||||||| t| |  S)aSort the dataset by a single column. Sorting a parallel dataset requires expensive shuffles and is generally not recommended. See ``set_index`` for implementation details. Parameters ---------- by: str or list[str] Column(s) to sort by. npartitions: int, None, or 'auto' The ideal number of output partitions. If None, use the same as the input. If 'auto' then decide by memory use. ascending: bool, optional Sort ascending vs. descending. Defaults to True. na_position: {'last', 'first'}, optional Puts NaNs at the beginning if 'first', puts NaN at the end if 'last'. Defaults to 'last'. sort_function: function, optional Sorting function to use when sorting underlying partitions. If None, defaults to ``M.sort_values`` (the partition library's implementation of ``sort_values``). sort_function_kwargs: dict, optional Additional keyword arguments to pass to the partition sorting function. By default, ``by``, ``ascending``, and ``na_position`` are provided. Examples -------- >>> df2 = df.sort_values('x') # doctest: +SKIP )rlastz,na_position must be either 'first' or 'last'c3>K|]}t|t ywrr)rbs rrz(DataFrame.sort_values.. s2!:a%%2rzuDataframes only support sorting by named columns which must be passed as a string or a list of strings. You passed %szLength of ascending=z != length of by=)r) rrrrrrr/rrrcrm) rr r ascending na_positionr sort_functionsort_function_kwargsr rrrs r sort_valueszDataFrame.sort_values sX / /KL L"d#B 2r2 2% "%b'*  )T*3y>SW3L4)5GB5IJ J $%n5   rc .tt|||S)aFilter dataframe with complex expression Blocked version of pd.DataFrame.query Parameters ---------- expr: str The query string to evaluate. You can refer to column names that are not valid Python variable names by surrounding them in backticks. Dask does not fully support referring to variables using the '@' character, use f-strings or the ``local_dict`` keyword argument instead. Notes ----- This is like the sequential version except that this will also happen in many threads. This may conflict with ``numexpr`` which will use multiple threads itself. We recommend that you set ``numexpr`` to use a single thread: .. code-block:: python import numexpr numexpr.set_num_threads(1) See also -------- pandas.DataFrame.query pandas.eval Examples -------- >>> import pandas as pd >>> import dask.dataframe as dd >>> df = pd.DataFrame({'x': [1, 2, 1, 2], ... 'y': [1, 2, 3, 4], ... 'z z': [4, 3, 2, 1]}) >>> ddf = dd.from_pandas(df, npartitions=2) Refer to column names directly: >>> ddf.query('y > x').compute() x y z z 2 1 3 2 3 2 4 1 Refer to column name using backticks: >>> ddf.query('`z z` > x').compute() x y z z 0 1 1 4 1 2 2 3 2 1 3 2 Refer to variable name using f-strings: >>> value = 1 >>> ddf.query(f'x == {value}').compute() x y z z 0 1 1 4 2 1 3 2 Refer to variable name using ``local_dict``: >>> ddf.query('x == @value', local_dict={"value": value}).compute() x y z z 0 1 1 4 2 1 3 2 )rrCrs rqueryzDataFrame.querysLeD$788rcg}|jD]]\}}|r4tjjj |j s<|j |j||_t|dS)Nrrrr) rirrrrrrurconcat)rrrrmodesrr!s rrzDataFrame.mode[sijjl KFAsBFFLL$A$A#))$L LL[I J Ke!$$rc@ttj||Sr)rr AddPrefixrprefixs r add_prefixzDataFrame.add_prefixddnnT6:;;rc@ttj||Sr)rr AddSuffixrsuffixs r add_suffixzDataFrame.add_suffixhr6rc t|||||S)a/ Create a spreadsheet-style pivot table as a DataFrame. Target ``columns`` must have category dtype to infer result's ``columns``. ``index``, ``columns``, ``values`` and ``aggfunc`` must be all scalar. Parameters ---------- values : scalar column to aggregate index : scalar column to be index columns : scalar column to be columns aggfunc : {'mean', 'sum', 'count'}, default 'mean' Returns ------- table : DataFrame ) pivot_table)rrrryaggfuncs rr=zDataFrame.pivot_tablels(4AArcddlm}||S)aLPurely integer-location based indexing for selection by position. Only indexing the column positions is supported. Trying to select row positions will raise a ValueError. See :ref:`dataframe.indexing` for more. Examples -------- >>> df.iloc[:, [2, 0, 1]] # doctest: +SKIP r) ILocIndexer)rr@)rr@s rrzDataFrame.ilocs C4  rcd| td|j|}t||||S)Nrrrlr)rexpr_clsrrrs r_comparison_opzDataFrame._comparison_ops8  %&:; ;""4(htUD9::rcF|jtj|||Sr)rDrLTFramerrrrs rltz DataFrame.lt""4<<tDDrcF|jtj|||Sr)rDrLEFramerGs rlez DataFrame.lerIrcF|jtj|||Sr)rDrGTFramerGs rgtz DataFrame.gtrIrcF|jtj|||Sr)rDrGEFramerGs rgez DataFrame.gerIrcF|jtj|||Sr)rDrNEFramerGs rnez DataFrame.nerIrcF|jtj|||Sr)rDrEQFramerGs reqz DataFrame.eqrIrc |}|j}|'t|jgdj}nt |r|g}|Dcgc] }t ||rt ||s|"}}|durOt |jrt |j }n#|!t|jjdv}t|s|dur|Sddl m }|t||||j\} }| jD cic]\} } | | j!} } } |t#|| |Scc}wcc} } w)aConvert columns of the DataFrame to category dtype. .. warning:: This method eagerly computes the categories of the chosen columns. Parameters ---------- columns : list, optional A list of column names to convert to categoricals. By default any column with an object dtype is converted to a categorical, and any unknown categoricals are made known. index : bool, optional Whether to categorize the index. By default, object indices are converted to categorical, and unknown categorical indices are made known. Set True to always categorize the index, False to never. split_every : int, optional Group partitions into groups of this size while performing a tree-reduction. If set to False, no tree-reduction will be used. kwargs Keyword arguments are passed on to compute. )rVstringcategoryF)rVrZrr)rrr)rrrrrrqrvrrrr$dask.dataframe.dask_expr._collectionrr9rrir*r8) rrrrrrwrQrr categoriesrr{s r categorizezDataFrame.categorizesN*xx ?4--.NOWWXG w iG  (a16J4PQ76S    #DJJ/0<<DJJ,,-1EE7|IG+ $u+ V ')  E 6@5E5E5GHTQa(H Hjz5ABB7 0Is %EE c |dk(r!ttj|||St|j Dcgc]'\}}|j ||j |)c}}Scc}}w)Nr)rrr.)rrNUniqueColumnsr/rinuniquer)rrrrrr!s rrazDataFrame.nuniquesp 19!$"5"5df"UV V&*ZZ\!cKKv;KGQQRVW s,A3 cgd}||vr tdtt|jj |||}|dk(r@t |t r td|jtj||d||S|r2|jdtjtjg }n|}g}|jD]'\} } |j| j || )t|d kDr4t |d t r!t#|||j$j&St)|d S) aApproximate row-wise and precise column-wise quantiles of DataFrame Parameters ---------- q : list/array of floats, default 0.5 (50%) Iterable of numbers ranging from 0 to 1 for the desired quantiles axis : {0, 1, 'index', 'columns'} (default 0) 0 or 'index' for row-wise, 1 or 'columns' for column-wise method : {'default', 'tdigest', 'dask'}, optional What method to use. By default will use dask's internal custom algorithm (``'dask'``). If set to ``'tdigest'`` will use tdigest for floats and ints and fallback to the ``'dask'`` otherwise. defaultr=tdigest1method can only be 'default', 'dask' or 'tdigest')qrrrz+'q' must be scalar when axis=1 is specifiedF)rrQrr)r1)rgrrr)rrrrsrquantilerrrrrrr6 datetime64rirurr3 _from_scalarsrrr/) rrgrrrallowed_methodsrQr collectionsrr!s rrhzDataFrame.quantilesG9  (PQ Q $** % . .,T /   19!T" !NOO&& !&) '  &&2>>2=="A'EE kkm AFAs   s||a|? @ A { a J{1~v$F dEJJ4F4FG Gk**rcf|dk(s|jdk(r|j||Std)Nrr+Dask doesn't implement an exact median in all cases as this is hard to do in parallel. See the `median_approximate` method instead, which uses an approximate algorithm.rmedian_approximater)rrrs rmedianzDataFrame.median&s? 19((A-**<*P P! `  rcH|j|||jdS)aReturn the approximate median of the values over the requested axis. Parameters ---------- axis : {0, 1, "index", "columns"} (default 0) 0 or ``"index"`` for row-wise, 1 or ``"columns"`` for column-wise method : {'default', 'tdigest', 'dask'}, optional What method to use. By default will use Dask's internal custom algorithm (``"dask"``). If set to ``"tdigest"`` will use tdigest for floats and ints and fallback to the ``"dask"`` otherwise. )rrrN)rhr)rrrrs rrpzDataFrame.median_approximate/s*}}f< &, rc |||ztjtjtjg}|jj |j }t|dk(rg|jj }nP|dk(r$| td|jj }n'|jj ||j }|Dcgc]}||j|||} }t| dScc}w) Nrrr^z*exclude must be None when include is 'all'r0)r percentilespercentiles_methodrr) rrr6rirrrrrdescriber/) rrrtrurr1_includerr!statss rrvzDataFrame.describe?s ?w 2>>2==AHjj..x.@HHG7|q **,,  " !MNNjj((Gjj..w.PXXG   I  ''#5    e!$$ sDcJ||}tj||g|_|S)Nr)rrr4)rrr/s rpopz DataFrame.pop_s$4jYYtdV4  rc<|ddl}|j}tt|j ddg}t |j dk(ro|jt|jjjd|jdt|jt||yi}|r,|j|j|jd|r|jdd |d <tt!|j#t%|j'}|rpddl}|d }|d } |jt+||jd t |j dddlmt1fd|j Ddz} t1| d} |j3dj5| j5dd} |j3dj5| } t7t!|j | j&|j8Dcgc]6\}\}}}| j5||||8}}}}}|j;| j=dnt+|j dg}|j;|t?|j8jAjCtDcgc]}d|z }}|jdj5d jE||r1|d jG}|jd!tI|dt||ycc}}}}wcc}w)"z5 Concise summary of a Dask DataFrame Nrz ._collectionrz : 0 entriesr)rrT)rwrrrrzData columns (total z columns):) pprint_thingc3@K|]}t|ywr)r)rrr|s rrz!DataFrame.info..sCLO,Csrz # {{column:<{column_width}}} Non-Null Count Dtype --- {{underl:<{column_width}}} -------------- -----) column_widthColumnz------)r underlzP {{i:^3}} {{name:<{column_width}}} {{count}} non-null {{dtype}})rrrrrColumnsr)r\z%s(%d)z dtypes: {}, zmemory usage: )%sysstdoutrr rrrrurrrrrrrrrrrrytextwraprwpandas.io.formats.printingr|rWdedentrrhrrsplitrO value_countsrirrr)rbufverboserrlines computationsrrcountsspacerheadercolumn_templaterrrr column_infor dtype_counts memory_intr|s @rinfozDataFrame.infoes- ; **CT$Z((<= t||  ! LLD!1!12;;Q>Q>S5TUV   )E!'*F LLu- . LL/DLL0A/B*M N ?CdllCCaGEua=LE \2x9 'ooTTZTZ!-U[UO09 fmmT[[A0  ,A+eU  &&"1o%d+&u-&u- ' K  LLd+ ,(IFGK [!"()A)A)C)I)I)KQT"U HqL    \((<)@AB %n599;J LL>+j*A)B"E F#u5 " s ;N  Nc(|j|||Sr)rM)rrFrrs rrGz DataFrame.covsyylK@@rc*|j||||Sr)rP)rrrFrrs rcorrzDataFrame.corrszz&+|[IIrcD|jj|dS)NFrshow_dimensionsrrrrs rrzDataFrame.to_strings! **He*TTrc|jj|d}t|jDchc]}|jc}}t dj ||jt|dScc}w)NFrzdataframe.html.j2r)rrlayers)rrrrr rrenderr)rrrr"rs rrzDataFrame.to_htmlsv (((E(Rtyy{3!agg34/077"6<88  4sBc$t||||||S)N)id_vars value_varsvar_name value_name col_level)melt)rrrrrrs rrzDataFrame.melts# !!   rc J|j}|j}|j}t|dk(r't j ggt|z||}|St j |jDcgc]\}}t||c}}d}|Scc}}w)Nr)rrrrr) rrrrrrr/rir,)rrQrcols series_dfrrs rrzDataFrame._repr_dataszz$$|| t9> bTCJ%6ERI  ?Czz|Ltq!"1E2LSTIMs<B )TPandasrNNN rNNNFF_x_yFNNN)NrrrNN)TNNNr)rNNFT)NT)NNTNFrr(rr/rzNone | Literal[0, 1])NrNraise) TFNNTN?AF) rr$rr/r floatrrrur/) NTr#rNNrFN)r zstr | list[str]rr$r&zbool | list[bool]r'zLiteral['first', 'last']rrr(z-Callable[[pd.DataFrame], pd.DataFrame] | Noner)zMapping[str, Any] | Noner rrz bool | Nonerr )TFFrRrr%)?rFrd)rF)rrdFFNrdNN)NFF)rONFFrNNNriN)Vrr,r-r.rr___annotations__rr_partition_typer3rarrfrrirkrmrprtrvr{rrrrrrrrrr rrrr_rrrrrrrrxrIr rr8rlrr rrlrrrr!r*r,rr5r;r=rrDrHrLrOrRrUrXr^rarhrqrprvrzrrGrrrrrr9r:s@rrr{ s&)UJ"*llO II  ",,* ***! . ",,% % ",,= = "2V(,",,) )V",,B B  y v",,  -  - ^",,DH   ($ 1$ 0 X",,   ",,   ",,O O",,    I    6#r**41K +K Z",,#D   ",, 2G3G ",, M  M",,F F",, O  O0 ",,2 2 ",,J J"& %~  ~ ~ ~ ~ ~ F#''+06 %GK9=$)%)I I  I % I . I  I EI 7I I "I #I VF9P",,% %",,< <",,< <B,!! ; EEEEEE8Ct",,   3+j",,    ",,$ % %>",,  Ob",,A A",, J J",,U U",,   ",,    " rrcPeZdZUdZeZded<ejZ e dZ e dZ e dZ e dZdZd Zeejd Zfd Ze d Zej*d Ze dZe dZdZdFdZeejdGdZeejdHdZeejefdZdZdIdZdIdZ dIdZ!dIdZ"dIdZ#dIdZ$eejdddddefd Z%eejdJd!Z&eejdKd"Z'eejdKd#Z(eejdLd$Z)dMd%Z*eejdNd&Z+ dOd'Z,e-d()edd*d+Z.e/dPdQd,Z0eejd-Z1eejd.Z2eejdRd/Z3eejdFd0Z4eejd1Z5eejd2Z6eejd3Z7e8d4e9Z:e8d5e;Z<e8d6e=Z>dSd7Z?eejd8Z@dTd9ZAdUd:ZBeejd;ZCdVd<ZDeejdWd=ZEeejdXd>ZFeejdYd?ZGeej dZd@ZHe eejdAZIe eejdBZJeejd[dCZKdDZLeMreejdEZNxZOSxZOS)\rzSeries-like Expr Collection. The constructor takes the expression that represents the query as input. The class is not meant to be instantiated directly. Instead, use one of the IO connectors from Dask. r^r_c|jfS)z Return a tuple representing the dimensionality of the DataFrame. The number of rows is a Delayed result. The number of columns is a concrete integer. )rrs rraz Series.shapes |rc|jgSrrrs rrkz Series.axess |rcy)rcrrrs rrz Series.ndim rdrctSrr}rs rrzSeries._elemwiserrcHttt|}|j|j|jttt j dD]*}t|j|r|j|,t|S)N)catr) rrr rrrrrrremover)rraccessors rrzSeries.__dir__sq DJ   S^$%& #H4::x0" #Awrctd)NzJUsing 'in' to test for membership is not supported. Use the values insteadr)rrs rrmzSeries.__contains__s! X  rc#K|j}t|jD]+}|j|j }|Ed{-y7wr)r-rrrr)rrrrs rrpzSeries.__iter__#sN t''( A##A&..0ALL  sA AAAc~t|ts|jdk(rt||S|j |Sr)rrrrarrrr\r s rrzSeries.__getitem__*s8 c6 "d&6&6!&;7&s+ +xx}rcB|jjjSr)rrrrs rrz Series.name/syy###rcF|j|j|_yNr)rr4rrs rrz Series.name3s[[t[,22 rcB|jjjSr)rrrrs rrz Series.dtype7syy$$$rc@t|jjS)zNumber of bytes)rrrrs rrz Series.nbytes;sdii..//rc |jdd}||zD]}t|tjr|jdk(r-t|t t tttjtjtjfr{tcS|dk(r6|jtS|jdkDrtSt!|g|i|StSrrrs rrzSeries.__array_ufunc__@rrNct|trYt|dkDrKt|ddtjr|ddj dk(rd}nA|ddj }n. ddl}d|jddd}t|dt||||jS#t$rd}Y9wxYw) Nrrrrr4rz0 is not implemented for `dask.dataframe.Series`.rz) rrrrrrarrrrrrzrrs rrzSeries.__array_wrap__Ys gu %#g,*:'!*Q-4A9L9LPR9R 1 ++ , !-'--/!"4Q"7!8: &-OP -&t,U%diiPP  ,+  ,rc t|trotj|j|jsE| t j t |dttj||d||S|Htjtj|||d}t j t |dttj||||S)Nrr)rrrQTrr) rrrrrrr2rMapAlignrrrr)rrrrQs rrz Series.mapls c6 "&&tyy#((;<MM,tE"BC%MM$ PTU <<<tSI4PD MM,tE: ;dhht PTUVVrc p|j|}t|jj|||Sr)rlrrrrs rrz Series.clipzs.""4(diinnUE4@AArcBttj||SNr)rrrrs rto_framezSeries.to_framesdll4d;<> !C (=(=$>'$I II $I >it3 D MY$.I &DDKKMEZF dIvy+yRX   rcLt|jj||Sr)rrr)rrrs rrz Series.modesdiinnV[ABBrc0tt|||SN)rorrrrors rrzSeries.nlargestshtqkJKKrc0tt|||Srrrs rrzSeries.nsmallestsi{KLLrc0tt|||Srrrs rrzSeries.memory_usagerrcFt|}tt||||S)z Return Series of unique values in the object. Includes NA values. Returns ------- uniques : Series )rhrr\)rrrrs runiquez Series.uniques$7~FfT; >RSSrc|j||}|r0t|tr|j}|j S|j S)N)rr)rrrrrr)rrrruniqss rrazSeries.nuniquesF$$ $R %');;= :: rc jt|}|dur tdtt||||||S)NFr)rrrrr)rhrrrQ)rrrrrrs rrzSeries.drop_duplicatessH7~F 5=%&GH H )#'-    rrrrc|j||turGtjtj ||f|dd|}t jt|t|jj |g|d|i|S)aParallel version of pandas.Series.apply Parameters ---------- func : function Function to apply $META args : tuple Positional arguments to pass to function in addition to the value. Additional keyword arguments will be passed as keywords to the function. Returns ------- applied : Series or DataFrame if func returns a Series. Examples -------- >>> import dask.dataframe as dd >>> s = pd.Series(range(5), name='x') >>> ds = dd.from_pandas(s, npartitions=2) Apply a function elementwise across the Series, passing in extra arguments in ``args`` and ``kwargs``: >>> def myadd(x, a, b=1): ... return x + a + b >>> res = ds.apply(myadd, args=(2,), b=1.5) # doctest: +SKIP By default, dask tries to infer the output metadata by running your provided function on some fake data. This works well in many cases, but can sometimes be expensive, or even fail. To avoid this, you can manually specify the output metadata with the ``meta`` keyword. This can be specified in many forms, for more information see ``dask.dataframe.utils.make_meta``. Here we specify the output is a Series with name ``'x'``, and dtype ``float64``: >>> res = ds.apply(myadd, args=(2,), b=1.5, meta=('x', 'f8')) In the case where the metadata doesn't change, you can also pass in the object itself directly: >>> res = ds.apply(lambda x: x + 1, meta=ds) See Also -------- Series.map_partitions T)rrrQ) rlrIrrrr rrr2r)rrrQrrrs rr z Series.applysxh D! : <<xWdWPVWD MM,t, -odiioohSSDSFSTTrc\|dvrtd|d|rddi}|j||S|S)N)rrNrz for Seriesrrrrs rrlzSeries._validate_axis2sA ) )~dV;?@ @ 9@! H<<d+ +Krc|Srrrs rr zSeries.squeeze<s rc>ttj|Sr)rr DropnaSeriesrs rrz Series.dropna@sd//566rcFttj||||S)N)rr inclusive)rrBetween)rrrrs rbetweenzSeries.betweenDs! LLD K  rc|j|d}tj|j|js!ttj||||Sttj ||||Srj)rrrrCombineSeriesAlign CombineSeries)rrrrs rrzSeries.combineJsk,,UG<""499ejj9!''eT:F d00udJOPPrc>ttj|Sr)rr ExplodeSeriesrs rrzSeries.explodeSsd00677rc@ttj||Sr)rrAddPrefixSeriesr3s rr5zSeries.add_prefixWd224@AArc@ttj||Sr)rrAddSuffixSeriesr9s rr;zSeries.add_suffix[rrrdtrc0tt||||Sr)rrM)rrr rxs r_repartition_quantileszSeries._repartition_quantilescs {Hl K  rc "ddlm}|||fi|S)Nr) SeriesGroupBy)rr)rr rrs rr zSeries.groupbyhsCT2000rcBttj|||S)aAlter Series index labels or name Function / dict values must be unique (1-to-1). Labels not contained in a dict / Series will be left as-is. Extra labels listed don't throw an error. Alternatively, change ``Series.name`` with a scalar value. Parameters ---------- index : scalar, hashable sequence, dict-like or callable, optional If dict-like or callable, the transformation is applied to the index. Scalar or hashable sequence-like will alter the ``Series.name`` attribute. inplace : boolean, default False Whether to return a new Series or modify this one inplace. sorted_index : bool, default False If true, the output ``Series`` will have known divisions inferred from the input series and the transformation. Ignored for non-callable/dict-like ``index`` or when the input series has unknown divisions. Note that this may only be set to ``True`` if you know that the transformed index is monotonically increasing. Dask will check that transformed divisions are monotonic, but cannot check all the values between divisions, so incorrectly setting this can result in bugs. Returns ------- renamed : Series See Also -------- pandas.Series.rename )rr RenameSeries)rr sorted_indexs rrz Series.renamensFd//e\JKKrclt|dgd}||vr tdtt|||S)aApproximate quantiles of Series Parameters ---------- q : list/array of floats, default 0.5 (50%) Iterable of numbers ranging from 0 to 1 for the desired quantiles method : {'default', 'tdigest', 'dask'}, optional What method to use. By default will use dask's internal custom algorithm (``'dask'``). If set to ``'tdigest'`` will use tdigest for floats and ints and fallback to the ``'dask'`` otherwise. rhrcrf)rkrrrL)rrgrrks rrhzSeries.quantiles; j18  (PQ QnT1f=>>rcV|jdk(r|jStd)Nrrnrors rrqz Series.medians2   q **, ,! `  rc&|j|S)aReturn the approximate median of the values over the requested axis. Parameters ---------- method : {'default', 'tdigest', 'dask'}, optional What method to use. By default will use Dask's internal custom algorithm (``"dask"``). If set to ``"tdigest"`` will use tdigest for floats and ints and fallback to the ``"dask"`` otherwise. r)rh)rrs rrpzSeries.median_approximates}}F}++rct|ts tdt||gd}|j |d|dSNz%other must be a dask.dataframe.SeriesrrT)rL)rrrr/rM)rrrFrrws rrGz Series.covs?%(CD D T5M *ww{D+dwCCrct|ts tdt||gd}|j ||d|dSr)rrrr/rP)rrrrFrrws rrz Series.corrsA%(CD D T5M *xx T;txLLrct|ts td|j|dk(r||S|j ||S)Nzlag must be an integerrr)rr rrr)rlagrs rautocorrzSeries.autocorrsE#x(45 5yyKyXXtzz#KyXXrct|jrt|jr*t|jst |jrt t ||||St t||||Sr)rrrrrrr=r<)rrrtrurr1s rrvzSeries.describesn TZZ (!$**-#DJJ/&tzz2!k;@RS ""4kCUV rc*tt|Sr)rrTrs ris_monotonic_increasingzSeries.is_monotonic_increasing3D9::rc*tt|Sr)rrSrs ris_monotonic_decreasingzSeries.is_monotonic_decreasingrrcB|jj|S)Nrrrs rrzSeries.to_strings **H*==rcBt|j|jSr)r,rrrs rrzSeries._repr_datas T-A-ABBrc t|ts tdtt|Dcgc] \}}||g c}}t}t t j |g|Scc}}w)Nz&The caselist argument should be a list container)rrrr%rrCaseWhen)rcaselistrr{s r case_whenzSeries.case_whens[h- HIIG$A1aV$ATRSH!$--"@x"@A A%BsA*rrrrr))rNrNTN)TFT)FNTNrr(r)both)rNr#)rrd)rdr&)rONF)rFrr)Prr,r-r.rr_rrrrr3rarkrrrrmrrprrr5rrrrrrrIrrDrHrLrOrRrUrXrrrrrrrarrxr r8rlr rrrrr5r;r)r7rr;r rdrr r rrhrqrprGrrrvrr!rrr'r)r9r:s@rrrs&)UJ"*iiO  "))  $$ [[33%%00"2Q&")) W W"))BB"))&==L RRRRRR"))  @"))CC"))LL"))MM"))OO T"))   ,#r**417U+7Ur"))"))77"))  "))QQ"))88"))BB"))BB  3 4C . /B  /C "))11 #LJ?$"))   ,"))DD "))MM "))YY "))$ ,"));;"));;"))>>C bii  B ! Brr)addsubmuldivdividetruedivfloordivmodpowraddrsubrmulrdivrtruediv rfloordivrmodrpowceZdZUdZeZded<ejZ hdZ hdZ fdZ dZ dd Zeejdefd Zeejd g d efdZeejddZddZeejddZdZddZedZdZdZdZdZdZdZdZ xZ!S) rzIndex-like Expr Collection. The constructor takes the expression that represents the query as input. The class is not meant to be instantiated directly. Instead, use one of the IO connectors from Dask. r^r_>dayhourweekyearmonthminutesecondquarterweekday dayofweek dayofyear nanosecond weekofyear microsecond millisecond>codesrorderedas_known as_ordered as_unknownr] as_unorderedadd_categoriesset_categoriesremove_categoriesrename_categoriesreorder_categoriesremove_unused_categoriesc~t|jjtjr$||j vrt |j|S||jvrt |j|Stt|r$t|jjd|t|=|S)Nz object has no attribute )rrrrr_cat_attributesrr_dt_attributesr rrarXr rr_rs rr_zIndex.__getattr__Is tzz'')<)< =t+++488S) ) D'' '477C( ( 57C !>>**--FsgN w"3''rc"d|jdS)Nz)rrs rr#zIndex.__repr__Zs-dii[::rNcDtj||jSr)rrr)rr'rs rrzIndex.__array_wrap__]sxxDII..rcT|tttj|||Sry)rrr ToSeriesIndexrrrs rrzIndex.to_series`s(  % %d00UNOOrrrTcT|stttj|||Sry)rrr ToFrameIndexras rrzIndex.to_framefs%% %d//EMNNrc.tt||S)N)rw)rrWrvs rrzIndex.memory_usagels.t$?@@rcBttj|||Sr)rr ShiftIndex)rrrs rrz Index.shiftpsdoodGTBCCrc t|trntj|j|jsD| t j t |dttj|||||S|Htjtj|||d}t j t |dttj|||||S)a  Note that this method clears any known divisions. If your mapping function is monotonically increasing then use `is_monotonic` to apply the mapping function to the old divisions and assign the new divisions to the output. rrTr)rrrQ is_monotonic) rrrrrrr2r MapIndexAlignrrrr)rrrrQrhs rrz Index.mapss c6 "&&tyy#((;<MM,tE"BC%&&tS)T<P <<<tSI4PD MM,tE: ; HH#L   rcttt|}|j|j|jttt j |j|jt|jtjr|j|jt|Sr)rrr rrrrr[rrrrrZrrs rrz Index.__dir__s~ DJ   S^$% $$% djj""5"5 6 HHT)) *Awrc,tt||Sr)rrRrs rrz Index.countsj{;<>> 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'), Timestamp('2010-01-03 00:00:00'), Timestamp('2010-01-05 00:00:00'), Timestamp('2010-01-06 00:00:00')) >>> ddf = from_pandas(df.a, npartitions=3) # Works with Series too! >>> ddf.divisions # doctest: +NORMALIZE_WHITESPACE (Timestamp('2010-01-01 00:00:00'), Timestamp('2010-01-03 00:00:00'), Timestamp('2010-01-05 00:00:00'), Timestamp('2010-01-06 00:00:00')) 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 z;Exactly one of npartitions and chunksize must be specified.rz+Input must be a pandas DataFrame or Series.zIndex in passed data is non-numeric and contains nulls, which Dask does not entirely support. Consider passing `data.loc[~data.isna()]` instead.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.r) FromPandas)rrrr{)rr/rrrqrrirrrdask.dataframe.dask_expr.io.iorrrfrr{)rrrrrs rrrs~!8VWW  {2 T "EFF zz'A$**'M! A  z+s'C a    z)S'A a  :   %#$;$=   rcddlm}t||jrt |||Sddlm}|||||}tr-|jjdvrtj|}t|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 rN)rrQ) FromArray)roriginal_columnsrQOU) dask.arrayr'rr!rrrr{rkindrArrowStringConversionr)rrrrQrrrs r from_arrayr>spB#rxx sG$??8    F  SYY^^t%;++F3 & !!rc :ddlm}t||||||S)Nr) FromGraph)rrrr name_prefix)rrr)rrrrrrs rrNrNqs)8 #   rpandasrc|jDchc]}t|st|}}|rtd|dt |j |||||Scc}w)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) zPfrom_dict doesn't currently support Dask collections as inputs. Objects of type z were given in the input dict.)ryr#r rrr)rrrrr constructorr{collection_typess rrrs{P*.PA:LQ:OQPP! /00N P  dFE7; Qs A" A"cdddlm}|t|tr t |sd}|||||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.DataFrame.values: Reverse conversion dask.dataframe.DataFrame.to_records: Reverse conversion r)rN)rrrQ)rrrrr)rrrrQrs rrrs3Z2z'48W 1gU FFrFc ddlm}m}m}t |t s t |}||d<|r||d<|Ht|tD]4}|\}}}|dk(st |tttfr+tdt |tjs"t |t r|jdvrtt j"td kr t%d | t'd | d k7r t'd | | dk7r t'd| t'd| dk7r t'd| t'dt)||t+|||||||||t-|t |t  St)||fidt+|d|d|d|d|d|d|d| d| d| d| d | d!|d"||d#|d$t |t S)%u, Read a Parquet file into a Dask DataFrame This reads a directory of Parquet data into a Dask.dataframe, one file per partition. It selects the index among the sorted columns if any exist. .. note:: Dask automatically resizes partitions to ensure that each partition is of adequate size. The optimizer uses the ratio of selected columns to total columns to squash multiple files into one partition. Additionally, the Optimizer uses a minimum size per partition (default 75MB) to avoid too many small partitions. This configuration can be set with >>> dask.config.set({"dataframe.parquet.minimum-partition-size": "100MB"}) # doctest: +SKIP .. note:: Specifying ``filesystem="arrow"`` leverages a complete reimplementation of the Parquet reader that is solely based on PyArrow. It is significantly faster than the legacy implementation, but doesn't yet support all features. Parameters ---------- path : str or list Source directory for data, or path(s) to individual parquet files. 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. columns : str or list, default None Field name(s) to read in as columns in the output. By default all non-index fields will be read (as determined by the pandas parquet metadata, if present). Provide a single field name instead of a list to read in the data as a Series. filters : Union[List[Tuple[str, str, Any]], List[List[Tuple[str, str, Any]]]], default None List of filters to apply, like ``[[('col1', '==', 0), ...], ...]``. Using this argument will result in row-wise filtering of the final partitions. Predicates can be expressed in disjunctive normal form (DNF). This means that the inner-most tuple describes a single column predicate. These inner predicates are combined with an AND conjunction into a larger predicate. The outer-most list then combines all of the combined filters with an OR disjunction. Predicates can also be expressed as a ``List[Tuple]``. These are evaluated as an AND conjunction. To express OR in predicates, one must use the (preferred for "pyarrow") ``List[List[Tuple]]`` notation. index : str, list or False, default None Field name(s) to use as the output frame index. By default will be inferred from the pandas parquet file metadata, if present. Use ``False`` to read all fields as columns. categories : list or dict, default None For any fields listed here, if the parquet encoding is Dictionary, the column will be created with dtype category. Use only if it is guaranteed that the column is encoded as dictionary in all row-groups. If a list, assumes up to 2**16-1 labels; if a dict, specify the number of labels expected; if None, will load categories automatically for data written by dask, not otherwise. storage_options : dict, default None Key/value pairs to be passed on to the file-system backend, if any. Note that the default file-system backend can be configured with the ``filesystem`` argument, described below. open_file_options : dict, default None Key/value arguments to be passed along to ``AbstractFileSystem.open`` when each parquet data file is open for reading. Experimental (optimized) "precaching" for remote file systems (e.g. S3, GCS) can be enabled by adding ``{"method": "parquet"}`` under the ``"precache_options"`` key. Also, a custom file-open function can be used (instead of ``AbstractFileSystem.open``), by specifying the desired function under the ``"open_file_func"`` key. dtype_backend : {'numpy_nullable', 'pyarrow'}, defaults to NumPy backed DataFrames Which dtype_backend to use, e.g. whether a DataFrame should have NumPy arrays, nullable dtypes are used for all dtypes that have a nullable implementation when 'numpy_nullable' is set, pyarrow is used for all dtypes if 'pyarrow' is set. ``dtype_backend="pyarrow"`` requires ``pandas`` 1.5+. calculate_divisions : bool, default False Whether to use min/max statistics from the footer metadata (or global ``_metadata`` file) to calculate divisions for the output DataFrame collection. Divisions will not be calculated if statistics are missing. This option will be ignored if ``index`` is not specified and there is no physical index column specified in the custom "pandas" Parquet metadata. Note that ``calculate_divisions=True`` may be extremely slow when no global ``_metadata`` file is present, especially when reading from remote storage. Set this to ``True`` only when known divisions are needed for your workload (see :ref:`dataframe-design-partitions`). ignore_metadata_file : bool, default False Whether to ignore the global ``_metadata`` file (when one is present). If ``True``, or if the global ``_metadata`` file is missing, the parquet metadata may be gathered and processed in parallel. Parallel metadata processing is currently supported for ``ArrowDatasetEngine`` only. metadata_task_size : int, default configurable If parquet metadata is processed in parallel (see ``ignore_metadata_file`` description above), this argument can be used to specify the number of dataset files to be processed by each task in the Dask graph. If this argument is set to ``0``, parallel metadata processing will be disabled. The default values for local and remote filesystems can be specified with the "metadata-task-size-local" and "metadata-task-size-remote" config fields, respectively (see "dataframe.parquet"). split_row_groups : 'infer', 'adaptive', bool, or int, default 'infer' If True, then each output dataframe partition will correspond to a single parquet-file row-group. If False, each partition will correspond to a complete file. If a positive integer value is given, each dataframe partition will correspond to that number of parquet row-groups (or fewer). If 'adaptive', the metadata of each file will be used to ensure that every partition satisfies ``blocksize``. If 'infer' (the default), the uncompressed storage-size metadata in the first file will be used to automatically set ``split_row_groups`` to either 'adaptive' or ``False``. blocksize : int or str, default 'default' The desired size of each output ``DataFrame`` partition in terms of total (uncompressed) parquet storage space. This argument is currently used to set the default value of ``split_row_groups`` (using row-group metadata from a single file), and will be ignored if ``split_row_groups`` is not set to 'infer' or 'adaptive'. Default is 256 MiB. aggregate_files : bool or str, default None WARNING: Passing a string argument to ``aggregate_files`` will result in experimental behavior. This behavior may change in the future. Whether distinct file paths may be aggregated into the same output partition. This parameter is only used when `split_row_groups` is set to 'infer', 'adaptive' or to an integer >1. A setting of True means that any two file paths may be aggregated into the same output partition, while False means that inter-file aggregation is prohibited. For "hive-partitioned" datasets, a "partition"-column name can also be specified. In this case, we allow the aggregation of any two files sharing a file path up to, and including, the corresponding directory name. For example, if ``aggregate_files`` is set to ``"section"`` for the directory structure below, ``03.parquet`` and ``04.parquet`` may be aggregated together, but ``01.parquet`` and ``02.parquet`` cannot be. If, however, ``aggregate_files`` is set to ``"region"``, ``01.parquet`` may be aggregated with ``02.parquet``, and ``03.parquet`` may be aggregated with ``04.parquet``:: dataset-path/ ├── region=1/ │ ├── section=a/ │ │ └── 01.parquet │ ├── section=b/ │ └── └── 02.parquet └── region=2/ ├── section=a/ │ ├── 03.parquet └── └── 04.parquet Note that the default behavior of ``aggregate_files`` is ``False``. parquet_file_extension: str, tuple[str], or None, default (".parq", ".parquet", ".pq") A file extension or an iterable of extensions to use when discovering parquet files in a directory. Files that don't match these extensions will be ignored. This argument only applies when ``paths`` corresponds to a directory and no ``_metadata`` file is present (or ``ignore_metadata_file=True``). Passing in ``parquet_file_extension=None`` will treat all files in the directory as parquet files. The purpose of this argument is to ensure that the engine will ignore unsupported metadata files (like Spark's '_SUCCESS' and 'crc' files). It may be necessary to change this argument if the data files in your parquet dataset do not end in ".parq", ".parquet", or ".pq". filesystem: "fsspec", "arrow", or fsspec.AbstractFileSystem backend to use. dataset: dict, default None Dictionary of options to use when creating a ``pyarrow.dataset.Dataset`` object. These options may include a "filesystem" key to configure the desired file-system backend. However, the top-level ``filesystem`` argument will always take precedence. **Note**: The ``dataset`` options may include a "partitioning" key. However, since ``pyarrow.dataset.Partitioning`` objects cannot be serialized, the value can be a dict of key-word arguments for the ``pyarrow.dataset.partitioning`` API (e.g. ``dataset={"partitioning": {"flavor": "hive", "schema": ...}}``). Note that partitioned columns will not be converted to categorical dtypes when a custom partitioning schema is specified in this way. read: dict, default None Dictionary of options to pass through to ``engine.read_partitions`` using the ``read`` key-word argument. arrow_to_pandas: dict, default None Dictionary of options to use when converting from ``pyarrow.Table`` to a pandas ``DataFrame`` object. Only used by the "arrow" engine. **kwargs: dict (of dicts) Options to pass through to ``engine.read_partitions`` as stand-alone key-word arguments. Note that these options will be ignored by the engines defined in ``dask.dataframe``, but may be used by other custom implementations. Examples -------- >>> df = dd.read_parquet('s3://bucket/my-parquet-data') # doctest: +SKIP See Also -------- to_parquet pyarrow.parquet.ParquetDataset r)ReadParquetFSSpecReadParquetPyarrowFS_set_parquet_engine dtype_backendarrow_to_pandasr%inz2Value of 'in' filter must be a list, set or tuple.)arrowpyarrowz15.0.0z:pyarrow>=15.0.0 is required to use the pyarrow filesystem.zFmetadata_task_size is not supported when using the pyarrow filesystem.inferzDsplit_row_groups is not supported when using the pyarrow filesystem.rdz=blocksize is not supported when using the pyarrow filesystem.zCaggregate_files is not supported when using the pyarrow filesystem.z.parqz.parquetz.pqzJparquet_file_extension is not supported when using the pyarrow filesystem.z:engine is not supported when using the pyarrow filesystem.) rfiltersr]rcalculate_divisionsstorage_options filesystemignore_metadata_filerr{r_seriesrrr]rrrrmetadata_task_sizesplit_row_groups blocksizeaggregate_filesparquet_file_extensionrenginerr)rrrrrrrr%rrrrpa_fs FileSystemr parse_versionpa __version__rrrrgr{)rrrr]rrrrrrrrrrrrrrrrrfilterr!rr^s r read_parquetrsj dC d#+F?$3 !g6 VF!LCSTz*S3e2D"E TUU V :u//0 j# &    "6 6  (=+B BL   )%X  w &%V   Y)%;%O   &%U  "%A A%\   %L  (1%$7 /%%9 /(?(A"7C0   $   $W-  "    ,  !4 "6  2 .   , $: " 'v. ! "w,#  rc t|ts tdt|dk(r t dt|dk(r0|dk(r&t|dt r|dj S|dS|dvr t d|Dcgc]}t|ts t|n|!}}|dk(r8|Dcgc]-}t|jdkDst|t s,|/}}tt||||||g|Scc}wcc}w)a@Concatenate DataFrames along rows. - When axis=0 (default), concatenate DataFrames row-wise: - If all divisions are known and ordered, concatenate DataFrames keeping divisions. When divisions are not ordered, specifying interleave_partition=True allows concatenate divisions each by each. - If any of division is unknown, concatenate DataFrames resetting its division to unknown (None) - When axis=1, concatenate DataFrames column-wise: - Allowed if all divisions are known. - If any of division is unknown, it raises ValueError. Parameters ---------- dfs : list List of dask.DataFrames to be concatenated axis : {0, 1, 'index', 'columns'}, default 0 The axis to concatenate along join : {'inner', 'outer'}, default 'outer' How to handle indexes on other axis interleave_partitions : bool, default False Whether to concatenate DataFrames ignoring its order. If True, every divisions are concatenated each by each. ignore_unknown_divisions : bool, default False By default a warning is raised if any input has unknown divisions. Set to True to disable this warning. ignore_order : bool, default False Whether to ignore order when doing the union of categoricals. Notes ----- This differs in from ``pd.concat`` in the when concatenating Categoricals with different categories. Pandas currently coerces those to objects before concatenating. Coercing to objects is very expensive for large arrays, so dask preserves the Categoricals by taking the union of the categories. Examples -------- If all divisions are known and ordered, divisions are kept. >>> import dask.dataframe as dd >>> a # doctest: +SKIP dd.DataFrame >>> b # doctest: +SKIP dd.DataFrame >>> dd.concat([a, b]) # doctest: +SKIP dd.DataFrame Unable to concatenate if divisions are not ordered. >>> a # doctest: +SKIP dd.DataFrame >>> b # doctest: +SKIP dd.DataFrame >>> dd.concat([a, b]) # doctest: +SKIP ValueError: All inputs have known divisions which cannot be concatenated in order. Specify interleave_partitions=True to ignore order Specify interleave_partitions=True to ignore the division order. >>> dd.concat([a, b], interleave_partitions=True) # doctest: +SKIP dd.DataFrame If any of division is unknown, the result division will be unknown >>> a # doctest: +SKIP dd.DataFrame >>> b # doctest: +SKIP dd.DataFrame >>> dd.concat([a, b]) # doctest: +SKIP dd.DataFrame By default concatenating with unknown divisions will raise a warning. Set ``ignore_unknown_divisions=True`` to disable this: >>> dd.concat([a, b], ignore_unknown_divisions=True)# doctest: +SKIP dd.DataFrame Different categoricals are unioned >>> dd.concat([ ... dd.from_pandas(pd.Series(['a', 'b'], dtype='category'), 1), ... dd.from_pandas(pd.Series(['a', 'c'], dtype='category'), 1), ... ], interleave_partitions=True).dtype CategoricalDtype(categories=['a', 'b', 'c'], ordered=False, categories_dtype=object) z/dfs must be a list of DataFrames/Series objectsrzNo objects to concatenater)rrkz!'join' must be 'inner' or 'outer') rrrrrrrrrrrr:)dfsrrignore_unknown_divisions ignore_orderinterleave_partitionsrrws rr/r/sJ c4 IJJ 3x1}455 3x1} 19CFF3q6??$ $1v  %%<==NQ R*R";;r? C RC R qyQb3rzz?Q#6*R:PrQQ     $ !       SRs$D5-D#Dc Dttj|d|||||S)NT)rrrrrr)rrr)rrrrrrs rrrs-     rc \|||fD]} t| tst|s6|s4|s2|s0|s.|jDcgc]}|jvs|}}|sdx}}|r|s|s|x}}tj j j|rt|ts t|}tj j j|rt|ts t|}d}||vrtd|d|d|dk(ro|stfdt|Dr tdt|jttt|t| |}t|s)|r|r|j!||}d}d}t#|d }ts)|r|rj!|d}d}t#d t%j&sJ|r|rt)|||t+t-|||||||| t/| | | }|j&j0j2j&j0j2k7r0|j5|j&j0j2 S|Scc}w)NT)rrrkrleftsemiz+dask.dataframe.merge does not support how='z'.Options are: rrc3:K|]}|jvywrr)rrrs rrzmerge..s '(AU]] " rzMhow='leftsemi' does not support right_index=True or on columns from the indexrrr1) rorrrrrrr _npartitionsrr)rrrrrrrrrrrrgrrrr#r!rr0rrtrrKrmrrr)rrrorrrrrrrrrrrr supported_howrs ` rrrs '8 $( a #%' '( ghz+ CM -9#?)?! -   j # ,>$w-0DGJ4Q/ e $ (OOE(O4EHKEq1 U[[ )) )8D%(;   !#0@$ F  zz 1 1 6 66!! (:(:(?(?!@@ M=s J)J)c | dvr td|||||||| | | | | d }|| tdt|tjr1t|tjrtj||fi|S||| td|x}}||d<||d<|d=||fD]}t|t st d t|s t|d }t|s t|d }||| td |x|d <|d<|d=| | td| | tdddl m }t|||fi|S)N)backwardforwardnearestzLInvalid merge_asof direction. Choose from 'backward' 'forward', or 'nearest') rrrrrr left_byright_byr toleranceallow_exact_matches directionzCannot merge_asof on NonezSCan only pass argument 'on' OR 'left_on' and 'right_on', not a combination of both.rrrz7Dask collections not currently allowed in merge columnsrr1zSCan only pass argument 'by' OR 'left_by' and 'right_by', not a combination of both.rrr z;Must specify both left_on and right_on if one is specified.r) MergeAsof) rrrr merge_asofrrr#r$dask.dataframe.dask_expr._merge_asofrr)rrrrrrrr rrrrrrrrrs rrrs":: '   "2 F |u}455$ %*UBLL*I}}T53F33 ~  ("6'  (#y%zt x  a #%I  d #4Q/ e $Eq1 ~  ("6e 243yF:.t 8/VWWx/VWW> )D%:6: ;;r)rrQrrrc"ddlm}m} ddlm} d|vr t dt } t|}t|D]G\} } t| tstdt|  | jt| It| dk(r tdt| dkDr td | dhk(r td d }d }t#j$|j&j)d d x}r/d}|j*|j,u}|t.ur1|r/t!dt|| rt1j2dnd }|gn|}|in|}|r-|j5d d }t7| |||||||||| }nt7|||||||||}t9rt7t;j<|S|S#tt f$r/t| || <| jt|| YwxYw)aCreate a DataFrame collection from a custom function map. ``from_map`` is the preferred option when reading from data sources that are not natively supported by Dask or if the data source requires custom handling before handing things of to Dask DataFrames. Examples are things like binary files or other unstructured data that doesn't have an IO connector. ``from_map`` supports column projection by the optimizer. The optimizer tries to push column selections into the from_map call if the function supports a ``columns`` argument. Parameters ---------- func : callable Function used to create each partition. Column projection will be enabled if the function has a ``columns`` keyword argument. *iterables : Iterable objects Iterable objects to map to each output partition. All iterables must be the same length. This length determines the number of partitions in the output collection (only one element of each iterable will be passed to ``func`` for each partition). args : list or tuple, optional Positional arguments to broadcast to each output partition. Note that these arguments will always be passed to ``func`` after the ``iterables`` positional arguments. $META divisions : tuple, str, optional Partition boundaries along the index. For tuple, see https://docs.dask.org/en/latest/dataframe-design.html#partitions For string 'sorted' will compute the delayed values to find index values. Assumes that the indexes are mutually sorted. If None, then won't use index information label : str, optional String to use as the function-name label in the output collection-key names. token : str, optional String to use as the "token" in the output collection-key names. enforce_metadata : bool, default True Whether to enforce at runtime that the structure of the DataFrame produced by ``func`` actually matches the structure of ``meta``. This will rename and reorder columns for each partition, and will raise an error if this doesn't work, but it won't raise if dtypes don't match. **kwargs: Key-word arguments to broadcast to each output partition. These same arguments will be passed to ``func`` for every output partition. Examples -------- >>> 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: string The optimizer will identify a column selection that happens after from_map and push the columns argument into the actual map call to drop unnecessary columns as early as possible. >>> def map_function(x, columns=None): ... df = pd.DataFrame({"a": [1, 2], "b": x}) ... if columns is not None: ... df = df[columns] ... return df >>> dd.from_map(map_function, [1, 2])["b"].compute() 0 1 1 1 0 2 1 2 Name: b, dtype: int64 This API can also be used as an alternative to other file-based IO functions, like ``read_csv`` (which are already just ``from_map`` wrapper functions): >>> import pandas as pd >>> import dask.dataframe as dd >>> paths = ["0.csv", "1.csv", "2.csv"] >>> dd.from_map(pd.read_csv, 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 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.csv", "1.csv", "2.csv"] >>> file_sizes = [86400, 86400, 86400] >>> def func(path, row_offset): ... # Read parquet file and set RangeIndex offset ... df = pd.read_csv(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 r)FromMapFromMapProjectable)DataFrameIOFunctionr>z,dask_expr does not support a token argument.z2All elements of `iterables` must be Iterable, got z/`from_map` requires at least one Iterable inputrz)All `iterables` must have the same lengthz+All `iterables` must have a non-zero lengthFrNTzArgument `func` of `from_map` has a required `columns` parameter and not `meta` provided.Either provide `meta` yourself or make `columns` an optional argument.zdask_expr does not support the DataFrameIOFunction protocol for column projection. To enable column projection, please ensure that the signature of `func` includes a `columns=` keyword argument instead.)dask.dataframe.dask_expr.iorrdask.dataframe.io.utilsrrrrrhrrrr r,rrXrrr parametersr0rdrfrIrrrzrr{rr)rrrQrrr iterablesrrrrrriterableallow_projectioncolumns_arg_requiredparamrrs rfrom_mapres?TH;&!"PQQeGYI + + 8(H-DT(^DTU  + KKH & + 7|qJKK W DEE1#~FGG !!$'2266y$GGuG$}} ; : "6Y  D- .  > !24D>RvF**Y- $          d88@AA MC * +>IaL KKIaL) * +s,G:H Hct|tr|j||St|s t |r(t t t||tStdt|d)aRepartition dataframe along new divisions Dask.DataFrame objects are partitioned along their index. Often when multiple dataframes interact we need to align these partitionings. The ``repartition`` function constructs a new DataFrame object holding the same data but partitioned on different values. It does this by performing a sequence of ``loc`` and ``concat`` calls to split and merge the previous generation of partitions. Parameters ---------- divisions : list List of partitions to be used force : bool, default False Allows the expansion of the existing divisions. If False then the new divisions lower and upper bounds must be the same as the old divisions. Examples -------- >>> df = df.repartition([0, 5, 10, 20]) # doctest: +SKIP Also works on Pandas objects >>> ddf = dd.repartition(df, [0, 5, 10, 20]) # doctest: +SKIP )rr)rr{z#repartition is not implemented for r) rrr4r0r1rrnrfr{rr )rwrrs rr4r4Csr:"i ~~ ~?? 2 ."4 R #(?(A   "$GRzQR"STTrc t|r|jjvr tdt|r|jjvr tdt j j|s tdt j|s tdt|r|jjvs*t|stfd|Ds tdgd}t|r||vr$tdd jd |Dztt|||| S) a Create a spreadsheet-style pivot table as a DataFrame. Target ``columns`` must have category dtype to infer result's ``columns``. ``index``, ``columns``, and ``aggfunc`` must be all scalar. ``values`` can be scalar or list-like. Parameters ---------- df : DataFrame index : scalar column to be index columns : scalar column to be columns values : scalar or list(scalar) column(s) to aggregate aggfunc : {'mean', 'sum', 'count', 'first', 'last'}, default 'mean' Returns ------- table : DataFrame See Also -------- pandas.DataFrame.pivot_table z.'index' must be the name of an existing columnz0'columns' must be the name of an existing columnz 'columns' must be category dtypez$'columns' must have known categoriesc3hK|])}t|xr|jjv+ywr)rrr)rrrws rrzpivot_table..s,G1 ! 6bhh&6&6!66Gs/2z4'values' must refer to an existing column or columns)rRrrrr#zaggfunc must be either rc3(K|] }d|d yw)rNr)rrs rrzpivot_table..s1WqAaS(1Ws)rrryr>) rrrrrrqrvr^rrr[)rwrrryr>available_aggfuncss` rr=r=ns'4 U uBHH,<,<<IJJ W 0@0@!@KLL  ' '(9 :;<<  1 2?@@ & bhh&& & GG GOPPB W 0B!B % 1WDV1W(W W   2UGFGT rdowncastrc|dvr tdt|r1| tdttj d|||St |rIttt|j|j||j|St|rtt||||Std ) z Return type depends on input. Delayed if scalar, otherwise same as input. For errors, only "raise" and "coerce" are allowed. )rcoercezinvalid error value specifiedz/``meta`` is not allowed when input is a scalar.T)pure)rrr)rrrrQzEarg must be a list, tuple, dask.array.Array, or dask.dataframe.Series)r pd_is_scalarrr}r to_numericrrrFrrrrr1r)rrrrQs rrrs ((899C  LM M0wr}}40VhWWC 3&&syy1&8   -T- "  # c C M   O rc |jdrddini}t|g\}|*t|trzt |j gfi|}|j |_tr|jt |jt|jj}nt|st|s tdt!|t#j$di|g}|j&j)|j&j*|_|j&j |j&_n t-|}|j/ddt1t3|||S)NutctzzSdask.dataframe.to_datetime does not support non-index-able arguments (like scalars)infer_datetime_format)rrrQ)2000)r0rjrrr DatetimeIndexrreas_unit to_datetimersrrr0r1rrzrrrrrrrrzrrE)rrQrtz_kwargs rrrs3 & 5 1e}rH  &FS | c5 !6#C(66rFXFDDI||$S)55mCII6NOTT$C(N3,?%:  0*3/1Q1Q0RSD**399??;DJ!iinnDJJO JJ&- *3vDI JJrcft|ts tdtt |||S)Nzarg must be a Series)rrr)rrrrrG)rrrs r to_timedeltars, c6 ".// +Cd6J KKrc.tt||g|Sr)rro)scalarsrQnamess rrjrjs +dENr)rrr|rHrirur rBr MapPartitionsrzrrr)rrQrrrrrrrrr+ newkwargsdelayed_kwargsrr{dexprnew_exprs rrrs"d"##FJ Kz!W5LO1 < KD KIN 1 a ! OE  ! !% ("5#6#6#8#;rct|trtj|}t|trtj|}t|tj st|tj r]t|t r!|jjj} n|jj} t| s@tdt|tr|dk\rt|tr|dk\s tdt|gd}t| } |r|g| z} | Dcgc]}t|t s|} }t| dkDrKt!j"| s9t%t!j&||||||||| j)dd| g | St!j*||||||||| j)dd| g | } t%| Scc}w)a] Apply a function to each partition, sharing rows with adjacent partitions. Parameters ---------- func : function The function applied to each partition. If this function accepts the special ``partition_info`` keyword argument, it will receive information on the partition's relative location within the dataframe. df: dd.DataFrame, dd.Series args, kwargs : Positional and keyword arguments to pass to the function. Positional arguments are computed on a per-partition basis, while keyword arguments are shared across all partitions. The partition itself will be the first positional argument, with all other arguments passed *after*. Arguments can be ``Scalar``, ``Delayed``, or regular Python objects. DataFrame-like args (both dask and pandas) will be repartitioned to align (if necessary) before applying the function; see ``align_dataframes`` to control this behavior. enforce_metadata : bool, default True Whether to enforce at runtime that the structure of the DataFrame produced by ``func`` actually matches the structure of ``meta``. This will rename and reorder columns for each partition, and will raise an error if this doesn't work, but it won't raise if dtypes don't match. before : int, timedelta or string timedelta The rows to prepend to partition ``i`` from the end of partition ``i - 1``. after : int, timedelta or string timedelta The rows to append to partition ``i`` from the beginning of partition ``i + 1``. transform_divisions : bool, default True Whether to apply the function onto the divisions and apply those transformed divisions to the output. align_dataframes : bool, default True Whether to repartition DataFrame- or Series-like args (both dask and pandas) so their divisions align before applying the function. This requires all inputs to have known divisions. Single-partition inputs will be split into multiple partitions. If False, all inputs must have either the same number of partitions or a single partition. Single-partition inputs will be broadcast to every partition of multi-partition inputs. $META See Also -------- dd.DataFrame.map_overlap zMMust have a `DatetimeIndex` when using string offset for `before` and `after`rz*before and after must be positive integersrr>N)rrrrrrrrrrrrr rrjrrrrMapOverlapAlignrz MapOverlap)rrwrrrQrrrrrrrrrs rrrGs@&#(%&&(,,-E8CUCU1V b) $HH33AAMHH22M&}5+  68$ aK uh ' QJEFF RD !! $B d #DdTk=b:b)#s 7G' G'ctt|tr|jSt|jSr)rrrqr)rs rrqrqs,#y!xxz3$$&&r)rQr/rc Rt|}|Dcgc]}t|ts|}}t|dkst j |r't t j|d||||g|}n%t t j|d|||g|}t||Scc}w)aElementwise operation for Dask dataframes Parameters ---------- op: callable Function to apply across input dataframes *args: DataFrames, Series, Scalars, Arrays, The arguments of the operation meta: pd.DataFrame, pd.Series (optional) Valid metadata for the operation. Will evaluate on a small piece of data if not provided. transform_divisions: boolean If the input is a ``dask.dataframe.Index`` we normally will also apply the function onto the divisions and apply those transformed divisions to the output. You can pass ``transform_divisions=False`` to override this behavior out : ``dask.array`` or ``None`` If out is a dask.DataFrame, dask.Series or dask.Scalar then this overwrites the contents of it with the result **kwargs: scalars Examples -------- >>> elemwise(operator.add, df.x, df.y) # doctest: +SKIP rr) rjrrrrrr UFuncElemwise UFuncAlign handle_out) rrQr/rrrrwrrs rr~r~s6 d #D :" 2y 92 :C : 3x1}++S1   s1vr41Df Tt T  AD& P4 PQ c6 "" ;s B$B$c t|tr/t|dk(r|d}nt|dkDr tdd}|N|j|jk7r5t dt t|dt t|t|trtt|jt|jk7rItdt t|jdt t|jt|tttfr|j|_ y|8dtt|d tt|d }t||S) a1Handle out parameters If out is a dask.DataFrame, dask.Series or dask.Scalar then this overwrites the contents of it with the result. The method replaces the expression of the out parameter with the result from this operation to perform something akin to an inplace modification. rrz4The `out` parameter with length > 1 is not supportedNz7Mismatched types between result and out parameter. out=z , result=z?Mismatched columns count between result and out parameter. out=z8The out parameter is not fully supported. Received type z , expected  )rrrrr rrr rrrrr3r4r)r/rrs rrrs6#u s8q=a&C X\%F C 3==F,<,<<#&tCy>3tF|3D F  #y! s{{ s6>>2 2'*3s{{+;'VX  # 623LL  c#f&   "#&& rc |jtj|}|jtj|}|jt|}t |||\}}}|j j}|j j}t||Dcgc] \}}|dk7s |}}}t||Dcgc] \}}|dk7s |}}}t||k7st||k7rFtd|jxsddd|jxsddtt||||sdtdt|d d |d d DrBtjd |jxsdd tt|||t t#j|}|s|||fS|||fScc}}wcc}}w)zFor a given column, compute the min, max, and len of each partition. And make sure that the partitions are sorted relative to each other. NOTE: this does not guarantee that every partition is internally sorted. rrz'Partitions are not sorted ascending by z the indexz. z.In your dataset the (min, max, len) values of z for each partition are: c3,K|] \}}||kywrr)rr+r%s rrz+_compute_partition_stats..?s!1aQ!srNrzPartitions have overlapping values, so divisions are non-unique. Use `set_index(sorted=True)` with no `divisions` to allow dask to fix the overlap. In your dataset the (min, max, len) values of z for each partition are : )rrrrWrrrrrrOrrrrrrrr) r rrrrmrnon_empty_minsnon_empty_maxess rrr&s  V 4D  ! !!%%f ! 5E  6 2DeT2D% ::<   D KKM "E),T4HIAvFaKaHNH*-eT*:JYQfkqJOJ~.0 / "o 55fkk6P[5QQS T>$ D eT""663IJs6 GG G(Gc,t|jSr)rpr)rs rget_parallel_type_framerPs QWW %%rrrr*)iPNNr)rr)NNNNNNNFFNrrdNrfsspecNN)rrkFFFrr) NNNFFNNNrNTrr#r)rNN)Nr)r rrr/rztuple[list, list, list[int]]) __future__rrrZrrrrrrrrnumbersr r typingr r r numpyrrrrr fsspec.utilsrpackaging.versionrrrrrrrrrrpandas.core.dtypes.commonrrrtlzrrr'rrr=dask.dataframe.methods dataframerrrdask._collectionsr dask._exprrdask._task_specrr r! dask.baser"r#r$ dask.corer%dask.dataframe._compatr&r'r(dask.dataframe.accessorr)dask.dataframe.corer*r+r,r-r.r/r0r1r2dask.dataframe.dask_exprr4rr6%dask.dataframe.dask_expr._categoricalr7r8r9 dask.dataframe.dask_expr._concatr:"dask.dataframe.dask_expr._datetimer;"dask.dataframe.dask_expr._describer<r=dask.dataframe.dask_expr._exprr>r?r@rArBrCrDrErFrGrHrIdask.dataframe.dask_expr._mergerJrK"dask.dataframe.dask_expr._quantilerL#dask.dataframe.dask_expr._quantilesrM$dask.dataframe.dask_expr._reductionsrNrOrPrQrRrSrTrUrVrWrXrYrZr[r\r]%dask.dataframe.dask_expr._repartitionr^r_!dask.dataframe.dask_expr._shuffler`rarbrc&dask.dataframe.dask_expr._str_accessorrddask.dataframe.dask_expr._utilrerfrgrhrirjrkrlrmrrnrodask.dataframe.dispatchrprqrrrsdask.dataframe.multirtdask.dataframe.utilsrurvrwrxryrzr{ dask.delayedr|r} dask.tokenizer~ dask.utilsrrrrrrrrrrrr dask.widgetsrrrrrrrrrsetattr partialmethodrrrrrr3r-rrrNregister_inplacerrrr/rrrrr4r=rrrrjrrrqr~rrregisterrrrrr5s-"AA$))'4#UU61>)((),%)LLOO2   3J 4?R    A=D$O B   I 5   *1    &$( G2\~4  ]$ ]$@ID  JB@ Ir2922=RHIA JD PB  Ir29223F2NO Pu up+KBYKB\ DD&y$'-- ' It7iHIvt$*d* $ FD4T6BC/D4WIFWItDYDDE`F0"f .--h7    181h1Gh.--h7   7 #o8oh " @J   (rxx    YYxr}}     R<R U$U$p ~$~$B'(Tt##L-b+0'7 '7#''7!'7TI&&'&r