CdWddlmZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z ddl mZmZddlZddlZddlZddlmZddlmZddlmZmZmZmZddlm Z dd l!m"Z"m#Z#m$Z$m%Z%m&Z&dd l'm(Z(m)Z)dd l*m+Z+dd l,m-Z-dd l.m/Z/m0Z0ddl1m2Z2ddl3m4Z4m5Z5m6Z6ej7e8Z9 d^d_d!Z: d`dad-Z; dbdcd7Z< ddded<Z= dfd=Z> dgd>Z? dhd?Z@Gd@dAZAdidBZBdCZC djdDZDdEZEdFZFdGZGdHZHdkdIZIdJZJdKZKdLZLejMdMZNdNZOdOZPdPZQdQZRdRZSdSZT dldmdWZUdndod[ZVdpd\ZW dqdrd]ZXdS)s) annotationsN)CallableMappingSequence)AnyLiteral)is_numeric_dtype)config)computecompute_as_if_collectionis_dask_collectiontokenize)methods) DataFrameSeries_Framemap_partitions new_dd_object)group_split_dispatchhash_object_dispatch)UNKNOWN_CATEGORIES)HighLevelGraph) ShuffleLayerSimpleShuffleLayer)sizeof)Mdigitget_default_shuffle_algorithm?ATdfr partition_colr repartitionbool npartitionsintupsamplefloatpartition_size ascendingreturntuple[list, list, list, bool]c f |rtng}||}tj} tj} t ||| | \}}} } nu#t$rh} tj sMtfdDr ddj dfnd\} } td| dj d | d | | d } ~ wwxYwtj|}|s|r(t!|}t t#j||z d }t |j}|j} t+jt+jd |d z |d zt+jd |d z ||}n#tt2f$rVt+jd |d z |d zt6}|j|}YnawxYw|j}t;|jd |d z |j|d z d z}| d} | d} tAj tj!r1j }| |} | |} |  s&|  rd}n | j}|r| jd |d z n| jd d "d}|r| jd d n| jd |d z "d}| | #|koU| | #|ko||k}|| | |fS)zO Utility function to calculate divisions for calls to `map_partitions` )r'c3DK|]}j|jkVdSN_name).0cr!r"s 6lib/python3.11/site-packages/dask/dataframe/shuffle.py z'_calculate_divisions..;s1FFa}*bek9FFFFFFcolumnz`.dropna(subset=['z'])`)seriesz`.loc[series[~series.isna()]]`z-Divisions calculation failed for non-numeric z 'z}'. This is probably due to the presence of nulls, which Dask does not entirely support in the index. We suggest you try with .Nr)xxpfpbfillmethodFTdrop)r*)$rr_repartition_quantilesrminmaxr TypeErrorr dtypeanynameNotImplementedErrorpdisnaallsummathceilr%sizenpinterplinspacetolist ValueErrorastyper&iloclistuniquefillna isinstanceCategoricalDtype reset_index sort_values)r!r"r#r%r'r)r*sizes divisionsminsmaxeseobjsuggested_methodempty_dataframe_detectedtotalnindexesrG presortedmaxes2mins2s`` r4_calculate_divisionsrn s*5 >s-7B DA#C<<DA,G--A$IIlastbystr | list[str]int | Literal['auto'] | Nonebool | list[bool] na_position"Literal['first'] | Literal['last'] sort_function-Callable[[pd.DataFrame], pd.DataFrame] | Nonesort_function_kwargsMapping[str, Any] | Nonec |dvrtdt|ts|g}td|Drt dt |z|||d} | t j}|| ||j dkr|j |fi| S|dkrd } td |j }n ||j }d } ||d } t|tskt|tr7t|dkr$t|d tr |d }nt d t |t|| | ||||\} }}}t| dkr"|dj |fi| S|r||j kr|j |fi| St!||d | ||d }|j |fi| }|S)z'See DataFrame.sort_values for docstring)firstroz,na_position must be either 'first' or 'last'c3BK|]}t|t VdSr/)r\str)r2bs r4r5zsort_values..s/ . .az!S!! ! . . . . . .r6zuDataframes only support sorting by named columns which must be passed as a string or a list of strings. You passed %s)rpr*rtNr:autoTdFrzHDask currently only supports a single boolean for ascending. You passed r%)r*rt duplicates)rVr\rYrHrJr}rr_updater%rrEr$lenrnr#rearrange_by_divisions)r!rpr%r*rtr'r)rvrxkwargs sort_kwargsr# sort_by_colrarbrcrks r4r_r_ws+++GHHH b$  T . .2 . . ... ! !"gg &   "K  '/000 ~ r >>+>>>f #r~..  .K RU)K i & &  y$ ' ' I!##9Q<..$"! II%k[^_h[i[ikk )= Kk8^Y))%ItUI 9~~;r~~!~,,;   (   ?[BN22 r >>+>>>   1    B  = 8 8K 8 8B Ir6Findex str | Seriesshuffle str | Noner rBraSequence | Nonec  |dkrd} td|j}n ||j}d} t|ts ||} n|} |dt || | |||\}} } }|rI||jkr>| | dgz}t ||||}|tjSt|||f|||d| S) z"See _Frame.set_index for docstringrTrNF)rBra)rrBr ) rEr%r\rrnset_sorted_indexrr sort_index set_partition)r!rr%rr rBr'rar)rr#index2rbrcrkresults r4 set_indexrs f #r~..  .K eV $ $E,@  [(N- - ) 4   766b {*I%b%diPPPF((66 6  E9 &-D'  MS  r6 r max_branch bool | Nonec &|jdg}t|trt |}t|t s||j}n|j}tj| r?tj j |r|j|}n_t|tj r)t|jvr|j|}n|j||}t|t s:||t"||} || } n4|t"||} || |} t'| d|t)|dz ||d } t|t s)| t*|||jj } n-| t.|j||jj } t3j|}td |D| _| t8jS) a'Group DataFrame by index Sets a new index and partitions data along that index according to divisions. Divisions are often found by computing approximate quantiles. The function ``set_index`` will do both of these steps. Parameters ---------- df: DataFrame/Series Data that we want to re-partition index: string or Series Column to become the new index divisions: list Values to form new divisions between partitions drop: bool, default True Whether to delete columns to be used as the new index shuffle: str (optional) Either 'disk' for an on-disk shuffle or 'tasks' to use the task scheduling framework. Use 'disk' if you are on a single machine and 'tasks' if you are on a distributed cluster. max_branch: int (optional) If using the task-based shuffle, the amount of splitting each partition undergoes. Increase this for fewer copies but more scheduler overhead. See Also -------- set_index shuffle partd rrG)rameta _partitions)r_indexrr:T)rr%rr ignore_index) index_namerB column_dtypec3ZK|]&}tj|s|n tjV'dSr/)rKrLrRnan)r2is r4r5z set_partition..Ys5MMa271::9!!26MMMMMMr6)_meta_constructor_slicedr\tuplerYrrGrKrLrHapitypesis_integer_dtyper]r categoriesrset_partitions_preassignrearrange_by_columnrset_index_post_scalarcolumnsset_index_post_seriesrIrrUrarr) r!rrarrBrr rrG partitionsdf2df3df4s r4rrsuP 8 ' ' , ,D)U##$OO eV $ $5   wy IBFL$A$A%$H$H IH00;; 5"-..I %"2 2 2H00;; H00%0HH eV $ $ >Y-- )$.  iiJi//)) )$*  iiJui==   NNQ&   C eV $ $    !) !    !z) !  y))IMM9MMMMMCM   al + ++r6c rtjj|ot | }|p t }|dkrt |ts|rht |tr|g}nt|}t|}|t|j z|krt|||||||St |ts'|rt|}| |}n$t|dr|}|t"|p|j|jdgd} || } |jjj| jj_t| d||||| } | d=| S) aGroup DataFrame by index Hash grouping of elements. After this operation all elements that have the same index will be in the same partition. Note that this requires full dataset read, serialization and shuffle. This is expensive. If possible you should avoid shuffles. This does not preserve a meaningful index/partitioning scheme. This is not deterministic if done in parallel. See Also -------- set_index set_partition shuffle_disk tasks)r%rrrr to_framerF)r%rtransform_divisionsrr)r%rrr r)rKrr is_list_liker rr\r}rYsetrrr_select_columns_or_indexhasattrrrpartitioning_indexr%rrrrrI) r!rrr%rrr list_likensetrrrs r4rr^s2 ))%00R9KE9R9R5RI8688G'z%55 eS ! ! GEEKKE5zz #bj// !T ) )&'%)  eV $ $ !  KKE++E22  # #!  %%12> X ) )1# . .! &J )) ) + +C8>.CIO  !   C M Jr6c`|j|}|s|}|jdg}||t||||} || } t | d|t|dz |} | d=| S)z:Shuffle dataframe so that column separates along divisionsr)rar*rtrrrr:)rr%r)rrdrop_duplicatesrrrrr) r!r7rarrr*rtrrrrrs r4rrs,,Y77I 0--// 8 ' ' , ,DF**  +J )) ) + +C   NNQ&    C M Jr6cl|p t}|!||jkr||}|dkrt||||S|dkr8t |||||}|r |jd|_|S|dkrd d lm}||||Std |z) Nrdisk)r rrTrAp2pr)rearrange_by_column_p2pzUnknown shuffle method %s) rr%r#rearrange_by_column_diskrearrange_by_column_tasksrr^distributed.shufflerrJ) r!colr%rrr rrrs r4rrs8688G ;#?#? ^^ ^ 4 4&'CgNNNN G  ' Z<     9 --4-88CI E  ??????&&r3 <<<!"="GHHHr6c&eZdZdZddZdZdZdS)maybe_buffered_partdz[ If serialized, will return non-buffered partd. Otherwise returns a buffered partd TNc|ptjdd|_||_tjdd|_dS)Ntemporary_directoryzdataframe.shuffle.compression)r gettempdirbuffer compression)selfrrs r4__init__zmaybe_buffered_partd.__init__s=I&*-BD"I"I  !:&EtLLr6cB|jrtd|jffStdfS)NFF)rr)rs r4 __reduce__zmaybe_buffered_partd.__reduce__s) < 4(5$,*?@ @((3 3r6c.ddl}tjd|j} |jrt |j|jnd}n:#t$r-}td |j|d}~wwxYw| |}|j j ||r ||}|jr;||||S||S)Nrz.partd)suffixdirzxNot able to import and load {} as compression algorithm.Please check if the library is installed and supported by Partd.)partdtempfilemkdtemprrgetattr compressedAttributeError ImportErrorformatFilefile cleanup_filesappendr PandasBlocksBufferDict)rargsrrpathpartd_compressionrdrs r4__call__zmaybe_buffered_partd.__call__s6 xT\BBB #($*:;;;      SSYSY$TT    zz$  ''---  +$$T**D ; ,%%ell5::<<&F&FGG G%%d++ +s#A A<(A77A<TN)__name__ __module__ __qualname____doc__rrrr6r4rrsTMMMM 444 ,,,,,r6rcnjt}tjj}d|zft fi}d|zfdt D}g}|rtj j ||} tj | g} t|g} tt| | \} } | i}tt!t|| }n|d|zt$t'|fi} d|zfdt)D}d d zz}t+j ||| |}tj ||} t-| j|S) zShuffle using local disk See Also -------- rearrange_by_column_tasks: Same function, but using tasks rather than partd Has a more informative docstring Nzzpartd-zshuffle-partition-c4i|]\}}|ft|fSr)shuffle_group_3)r2rkeyr7rIr%ps r4 z,rearrange_by_column_disk..7s>    As q OS&+qA   r6 dependencieszbarrier-zshuffle-collect-c8i|]}|ft|jfSr)collectr)r2r barrier_tokenr!rIrs r4rz,rearrange_by_column_disk..Ms9   @Aq GQ28];   r6r/r:)r%ruuiduuid1hexr enumerate __dask_keys__rmergedaskfrom_collectionssortedr rdictziprbarrierrYrangetoolzrr)r!r7r%r tokenalways_new_tokendsk1dsk2rgraphkeysppvaluesdsk3dsk4ralayerrrIrs``` @@@r4rr#sn R - -Ez||' % %'A $&&( )D "2 2D        0 0 2 233   D L $RWdD99/e2$OOO6$<< -iEE F2wCt f--..B!11M GT$ZZ0 1D  %D       EJ;EWEW   D;?+I KdD$ / /E  +D%l S S SE bh : ::r6c|S)z# A task that does nothing. r)r; cleanup_tokens r4_nooprXs  Hr6c |pd}p|j|kryt|}d|}p|jt||j|j|j}t j|||g}t|||jdgdzzS|j} ttj tj | tj |z dkr(ttj | dz zn| fdtzD} |j} t|| }tD]`} d| d|} t| | | | |j|j }t j| ||g}t|| |j|j}a| krt|}d |zfd t|D}d |z}tD]}t"|| zf|f|||f<t j|||g}t|||jdgdzz}n|}d | dzz|_|S) ar Order divisions of DataFrame so that all values within column(s) align This enacts a task-based shuffle. It contains most of the tricky logic around the complex network of tasks. Typically before this function is called a new column, ``"_partitions"`` has been added to the dataframe, containing the output partition number of every row. This function produces a new dataframe where every row is in the proper partition. It accomplishes this by splitting each input partition into several pieces, and then concatenating pieces from different input partitions into output partitions. If there are enough partitions then it does this work in stages to avoid scheduling overhead. Lets explain the motivation for this further. Imagine that we have 1000 input partitions and 1000 output partitions. In theory we could split each input into 1000 pieces, and then move the 1 000 000 resulting pieces around, and then concatenate them all into 1000 output groups. This would be fine, but the central scheduling overhead of 1 000 000 tasks would become a bottleneck. Instead we do this in stages so that we split each of the 1000 inputs into 30 pieces (we now have 30 000 pieces) move those around, concatenate back down to 1000, and then do the same process again. This has the same result as the full transfer, but now we've moved data twice (expensive) but done so with only 60 000 tasks (cheap). Note that the `column` input may correspond to a list of columns (rather than just a single column name). In this case, the `shuffle_group` and `shuffle_group_2` functions will use hashing to map each row to an output partition. This approach may require the same rows to be hased multiple times, but avoids the need to assign a new "_partitions" column. Parameters ---------- df: dask.dataframe.DataFrame column: str or list A column name on which we want to split, commonly ``"_partitions"`` which is assigned by functions upstream. This could also be a list of columns (in which case shuffle_group will create a hash array/column). max_branch: int The maximum number of splits per input partition. Defaults to 32. If there are more partitions than this then the shuffling will occur in stages in order to avoid creating npartitions**2 tasks Increasing this number increases scheduling overhead but decreases the number of full-dataset transfers that we have to make. npartitions: Optional[int] The desired number of output partitions Returns ------- df3: dask.dataframe.DataFrame See also -------- rearrange_by_column_disk: same operation, but uses partd rearrange_by_column: parent function that calls this or rearrange_by_column_disk shuffle_group: does the actual splitting per-partition rzsimple-shuffle-rNr:cdg|]+tfdtD,S)c3:K|]}t|VdSr/)r)r2jrks r4r5z7rearrange_by_column_tasks...s-::qE!QNN::::::r6)rr )r2rr!stagess @r4 z-rearrange_by_column_tasks..s@ U U Uqe:::::E&MM::::: U U Ur6zshuffle--zrepartition-group-c4i|]\}}|ft|fSr)shuffle_group_2)r2rr!r7rr%repartition_group_tokens r4rz-rearrange_by_column_tasks..sH    1%a ( +    r6zrepartition-get-r/)r%rrr1rrrrr&rOrPlogr rrarrshuffle_group_get)r!r7rr%rr shuffle_name shuffle_layerrriinputsnpartitions_origstage stage_name stage_layerdskrepartition_get_namergraph2rr!r'r"s ` `` @@@r4rr_s6v!rJ%r~*44V[11000 !3R^ *    N  H H  / -rd   UL"(TFkTUo> ? ?F zz  !F +,, - -  U U U U UE!V)DTDT U U UF~ RA . .EvFF///// "       H H   /  2$   5*bh E E;2B#B#B[))"6">        """2"2"4"455     2E9{##  A!(!.>*>?.C%q) * *  0 #RD    ("(TFkAo4N  #3a#78 Jr6cDt|dt|zS)a~ Computes a deterministic index mapping each record to a partition. Identical rows are mapped to the same partition. Parameters ---------- df : DataFrame/Series/Index npartitions : int The number of partitions to group into. Returns ------- partitions : ndarray An array of int64 values mapping each record to a partition. Fr)rr&)r!r%s r4rrs$" % 0 0 03{3C3C CCr6c$t|dS)Nr)rY)rs r4r r  sJJJ 1r6cddl}t||jr|j}nd}t||jr|j}nd}|rt j|ddSdS)z Cleanup the files in a partd.File dataset. Parameters ---------- p : partd.Interface File or Encode wrapping a file should be OK. keys: List Just for scheduling purposes, not actually used. rNT) ignore_errors)rr\Encoderrshutilrmtree)rrr maybe_filers r4cleanup_partd_filesr=s~LLL!U\""W  *ej)) 0 d$//////00r6ct|5||}t|dkr|n|cdddS#1swxYwYdS)z/Collect partitions from partd, yield dataframesrN)ensure_cleanup_on_exceptionrr)rpartrrress r4rr-s $Q ' '--eeDkk#hhllss------------------s,A  A A c |r||ddz }n*t|||dz dz }n#ttf$rt jt|d}|}||}|r$|||ddz ||<n3t||||dz dz ||<YnwxYw|rt|dz nd||dk|t|dz kz<|dkrt|dz nd||j<|S) Nright)sider:int32rrrro) searchsortedrrFrVrRemptynotnarLr)srar*rtrnot_nulldivisions_notnas r4rr4s  V"///@@1DJJY)*@*@*@*Q*QQTUUJ z "Xc!ffG444 7799#IOO$5$56  ,,Qx[w,GG!K x I!..q{.IIJ x  (.IQ Q:Y!1C#CDE9Dv8M8M#i..1"4"4STJqvvxx sAA B;DDct|si|fSt|tr|g}|r8|ddkr,||dtj}nHt |r||n|dt|ztj}|dz}t||||}||j ddfS)NrrFr5r:r) rr\r}rWrRrErr&rErrX)r!colsrnpartsindriresult2s r4r&r&Ps r772v $v Q=((ak  ** !T!9Dr G G G#f++ U &     A A"2sALIIIG BGBQBK r6c(|\}}||vr||S|Sr/r)g_headrgheads r4r)r)cs"GAtAvvt  r6cpt|tr|g}|r|ddkr||d}n5t|r||n|d}|r||kr|t|z}t j|dz}||z|d||zz|z}t||||S)aSplits dataframe into groups The group is determined by their final partition, and which stage we are in in the shuffle Parameters ---------- df: DataFrame cols: str or list Column name(s) on which to split the dataframe. If ``cols`` is not "_partitions", hashing will be used to determine target partition stage: int We shuffle dataframes with many partitions we in a few stages to avoid a quadratic number of tasks. This number corresponds to which stage we're in, starting from zero up to some small integer k: int Desired number of splits from this dataframe npartition: int Total number of output partitions for the full dataframe nfinal: int Total number of output partitions after repartitioning Returns ------- out: Dict[int, DataFrame] A dictionary mapping integers in {0..k} to dataframes such that the hash values of ``df[col]`` are well partitioned. rrFr5r)copyr)r\r}rr&rRmin_scalar_typerWr) r!rMr.r!r%rnfinalrOtyps r4 shuffle_grouprZks:$v $Q=((ak"t#;2d885III  $f ++F #C [1_ - -C   $ $Su $ 5 5E AA EC C F F FFr6c#K dVdS#t$rB |n*#t$rtdYnwxYwwxYw)zEnsure a partd.File is cleaned up. We have several tasks referring to a `partd.File` instance. We want to ensure that the file is cleaned up if and only if there's an exception in the tasks using the `partd.File`. Nz1ignoring exception in ensure_cleanup_on_exception) ExceptionrBlogger exception)rs r4r?r?s    R FFHHHH R R R   P Q Q Q Q Q R s) A*A$AAAAct|5||fdjD}||dddddS#1swxYwYdS)Nc<i|]}||Sr) get_group)r2rrSs r4rz#shuffle_group_3..s% 1 1 11Q A 1 1 1r6T)fsync)r?groupbygroupsr)r!rr%rdrSs @r4rrs $Q ' '  JJsOO 1 1 1 1 1 1 1 $                  sAAA"%A"c|dd||}|j||_|S)Nrr:axisrA)rBrrrWr!rrBrrs r4rrsF ''-a' ( ( 2 2:D 2 I IC+$$\22CK Jr6c|dddd}||j_|j||_|S)Nrr:rgrTrA)rBrrrIrrWris r4rrsO ''-a' ( ( 2 28$ 2 G GCCIN+$$\22CK Jr6cB||jvr||n|Sr/)rrBr!rs r4 drop_overlaprms""bh..2775>>>B6r6cX||jvr|j|gn|Sr/)rloc _constructorrls r4 get_overlaprqs*#rx//265'??R__5F5FFr6c dt|z dt|D}t|dkr>d} dfjdfi}t j |g}t | j|S fdt|D}t| }tdfz}fdtd tD} g} | D]} | t|| d z || ft| | d z f|| f| | d z f<|| || d zkr| d z| vrn| || tj| f| | f<g} t j |g}t | j|S) aEnsures that the upper bound on each partition of ddf (except the last) is exclusive This is accomplished by first removing empty partitions, then altering existing partitions as needed to include all the values for a particular index value in one partition. z fix-overlap-c$g|] \}}|dk |Srr)r2rlengths r4r#zfix_overlap..s!EEEF11r6r)NNrc.i|]\}}|fj|fSrr0)r2rdivddfrIs r4rzfix_overlap..s* L L L61cD!9sy#& L L Lr6rc>g|]}||dz k|S)r:rr2rrcrbs r4r#zfix_overlap..s/IIIQaE!a%L1H1Hq1H1H1Hr6r:)rrrr1rrrrrYrrr rrqrmrconcat) rxrbrclens non_emptiesrar1rddf_keysoverlapframesrrIs ``` @r4 fix_overlaprs HS$t<< ! MT1q5M Q<9QU+ + +A0@0@  hqk"""!.&1T1I  +D#SE J J JE ci ; ;;r6r7 allow_overlaptuple[list, list, list[int]]c |tj|}|tj|}|t|}t |||fi|\}}}|d}|d}dt||D}dt||D}t||kst||krCtd|j pdd|j pdd tt||||sztd t|d d |d d DrCtjd|j pdd tt|||t t#j|}|s|||fS|||fS)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. )rr>r?c$g|] \}}|dk |Srtrr2mrus r4r#z,_compute_partition_stats.. s!HHHIAvFaKKaKKKr6c$g|] \}}|dk |Srtrrs r4r#z,_compute_partition_stats.. s!JJJYQfkkqkkkr6z'Partitions are not sorted ascending by z the indexz.In your dataset the (min, max, len) values of z for each partition are : c3(K|] \}}||kVdSr/r)r2ar~s r4r5z+_compute_partition_stats..s;!!1aQ!!!!!!r6r:NrzPartitions 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 )rrrDrErr r[rUr r rVrIrYrHwarningswarn UserWarningr)r7rrrbrcr|non_empty_minsnon_empty_maxess r4_compute_partition_statsrsP  V 4 4D  ! !!%f ! 5 5E  6 2 2DeT<$  D 7eT""66r6r Any | Nonerc |||jn||}t|fddi|\}}}t||dfzS)NrFr)rrr)r!rrr7rbrc_s r4compute_divisionsr&sLRXX"S'F-fTTETVTTND% ;;%) %%r6c Xt|jfddi|\}tt|jdz krWt dfz|_t fdtdtDs|St||S)NrTr:rc3BK|]}||dz kVdS)r:Nrrzs r4r5z,compute_and_set_divisions..1s4HHq47eAEl*HHHHHHr6) rrrrar _divisionsrHr r)r!rr|rcrbs @@r4compute_and_set_divisionsr-s0XXXQWXXD% 4yyC %%)))d uRyl2 HHHHHE!SYY4G4GHHHHH I r4 - --r6c t|tr"|j|j|}n|j||}t t j||||dd}|s t |fi|St|t|jkrd}t|t||_|S)NrAF)rBralign_dataframesraeWhen 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)`) r\rrrrrrrrarVr)r!rrBrarrrmsgs r4rr7s%  4x!!%+D!99x!!%d!33     !F  (::6::: Y3r|,, , , P ooY''F Mr6)rr T)r!rr"rr#r$r%r&r'r(r)r(r*r$r+r,)NTrorr NN)r!rrprqr%rrr*rsrtrur'r(r)r(rvrwrxryr+r)NNFTrNr )r!rrrr%rrrrr r$rBr$r'r(rarr)r(r+r)rTNN)r!rrrrarrr&rBr$rrr rr+r)NNrFN)NNTroT)NNNNF)NF)rNF)Tror)r7rrr$r+rr/)r!rrrr+r)r!rr+rr) r!rrrrBr$rarr+r)Y __future__r contextlibloggingrOr:rrrcollections.abcrrrtypingrrnumpyrRpandasrKtlzrpandas.api.typesr rr dask.baser r r rdask.dataframerdask.dataframe.corerrrrrdask.dataframe.dispatchrrdask.dataframe.utilsrdask.highlevelgraphr dask.layersrr dask.sizeofr dask.utilsrrr getLoggerrr]rnr_rrrrrrrrrrr r=rrr&r)rZcontextmanagerr?rrrrmrqrrrrrrr6r4rs""""""  7777777777------UUUUUUUUUUUU""""""XXXXXXXXXXXXXXNNNNNNNN333333......88888888>>>>>>>>>>  8 $ $!T?T?T?T?T?t15#'6<!CG59RRRRRp15!%!&&&&&Zf,f,f,f,f,X  LLLLf ####R   I I I IF*,*,*,*,*,*,*,*,Z2;2;2;2;j   ?DRRRRtDDD(   0006---8   &+G+G+G\ &    777GGG0<0<0