LgVddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z m Z ddlZddlmZddlmZmZdd lmZdd lmZmZmZmZdd lmZdd lmZdd l m!Z!ddl"m#Z#m$Z$m%Z%m&Z&e rddl'Z(GddZ)GddeZ*Gdde*Z+Gdde*Z,Gdde*Z-Gdde*Z.e!j^e,dZ0e!j^e-dZ1e!j^e.dZ2GddeZ3d-d Z4d!Z5d"Z6Gd#d$eZ7Gd%d&e7Z8Gd'd(eZ9Gd)d*eZ:Gd+d,eZ;y). annotationsN) defaultdict)Callable)product) TYPE_CHECKINGAny)map)TaskTaskRef)tokenize) Blockwise BlockwiseDepBlockwiseDepDictblockwise_token)flatten)Layer)normalize_token)apply cached_cumsumconcreteinsertceZdZdZdZdZy)CallableLazyImportzFunction Wrapper for Lazy Importing. This Class should only be used when materializing a graph on a distributed scheduler. c||_yN) function_path)selfrs +lib/python3.12/site-packages/dask/layers.py__init__zCallableLazyImport.__init__'s *c>ddlm}||j|i|S)Nr) import_term)distributed.utilsr#r)rargskwargsr#s r__call__zCallableLazyImport.__call__*s"1.{4--.???r!N)__name__ __module__ __qualname____doc__r r'r!rrr s +@r!rcDeZdZUdZded<ded<dZded<d d Zd d Zy )ArrayBlockwiseDepzg Blockwise dep for array-likes, which only needs chunking information to compute its data. tuple[tuple[int, ...], ...]chunkstuple[int, ...] numblocksFboolproduces_taskscN||_td|D|_d|_y)Nc32K|]}t|ywr)len).0chunks r z-ArrayBlockwiseDep.__init__..Cs>ves5zvsF)r0tupler2r4)rr0s rr zArrayBlockwiseDep.__init__As# >v>>#r!ctd)Nz%Subclasses must implement __getitem__)NotImplementedErrorridxs r __getitem__zArrayBlockwiseDep.__getitem__Fs!"IJJr!Nr0r/r?r1)r(r)r*r+__annotations__r4r r@r,r!rr.r.7s* (' ND $ Kr!r.ceZdZdZddZy)ArrayChunkShapeDepz(Produce chunk shapes given a chunk indexcNtdt||jDS)Nc3,K|] \}}||ywrr,)r8ir9s rr:z1ArrayChunkShapeDep.__getitem__..NsD.C(!UU1X.Cs)r;zipr0r>s rr@zArrayChunkShapeDep.__getitem__MsDc#t{{.CDDDr!NrB)r(r)r*r+r@r,r!rrErEJs 2Er!rEc8eZdZUdZded<dfd ZddZxZS) ArraySliceDepz>Produce slice(s) into the full-sized array given a chunk indexr/startscRt||td|D|_y)Nc36K|]}t|dyw)T) initial_zeroN)r)r8cs rr:z)ArraySliceDep.__init__..XsPAM!$??)superr r;rL)rr0 __class__s rr zArraySliceDep.__init__Vs"  PPP r!crtdt||jD}td|DS)Nc3<K|]\}}||||dzfyw)Nr,)r8rHstarts rr:z,ArraySliceDep.__getitem__..[s(S=REU1XuQU|,=Rc36K|]}tg|dywrslice)r8ss rr:z,ArraySliceDep.__getitem__..\s2cU_A_t_crQ)r;rIrL)rr?locs rr@zArraySliceDep.__getitem__Zs.SSdkk=RSS2c222r!rAr?r;)r(r)r*r+rCr r@ __classcell__rSs@rrKrKQsH ''Q3r!rKceZdZddZy)ArrayBlockIdDepc<t|ts td|S)Nz&ArrayBlockIdDep requires a tuple index) isinstancer;r=r>s rr@zArrayBlockIdDep.__getitem__`s#u%%&NO O r!Nr^)r(r)r*r@r,r!rrbrb_sr!rbc(eZdZdfd ZddZxZS)ArrayValuesDepc2t||||_yr)rRr values)rr0rhrSs rr zArrayValuesDep.__init__hs   r!c |j|Sr)rhr>s rr@zArrayValuesDep.__getitem__ls{{3r!)r0r/rhznp.ndarray | dictr^)r(r)r*r r@r_r`s@rrfrfgs  r!rfcd|jfS)NrKr0deps rnormalize_array_slice_deprnps CJJ &&r!cd|jfS)Nrbrkrls rnormalize_array_block_id_deprpus cjj ((r!c4d|j|jfS)Nrf)r0rhrls rnormalize_array_values_deprrzs SZZ 33r!cdeZdZdZfdZdZedZdZdZ dZ dZ d Z d Z d d ZxZS) ArrayOverlapLayera\Simple HighLevelGraph array overlap layer. Lazily computed High-level graph layer for a array overlap operations. Parameters ---------- name : str Name of new output overlap array. array : Dask array axes: Mapping Axes dictionary indicating overlap in each dimension, e.g. ``{'0': 1, '1': 1}`` cvt|||_||_||_||_||_d|_yr)rRr nameaxesr0r2token _cached_keys)rrvrwr0r2rxrSs rr zArrayOverlapLayer.__init__s;    "  r!c"d|jdS)NzArrayOverlapLayer.keyssy d)CQw#i.(9>y~9NO9NA4'D.A4/9NOM9>in8MN8M1$.8MNMPNs A-A2)ryrvr0r2)rrr0rrvr2s @@@@r _dask_keyszArrayOverlapLayer._dask_keyssR    ($$ $"&))T[[$..fi &*V+F r!c|j}|j}|j}|j}d|jz}d|jz}|r t d}nddlm}ttt|} tjt| |} tj|t t| tt tj"t} i} i} | D]S}t%|f|z|}|dur|f|z|k7r || |f|z<+|f|z| |f|z<|t&| d|z| ff| |f|z<Utj(| | }|S) z/Construct graph for a simple overlap operation.zgetitem-zoverlap-zdask.array.core.concatenate3r) concatenate3)dimsrwFrr|)rwr0rvrrxrdask.array.corerlistr r7 functoolspartial_expand_keys_around_centertoolzpiperconcatfractional_slicermerge)r deserializingrwr0rv dask_keys getitem_name overlap_namerr expand_key2 interior_keysinterior_slicesoverlap_blocksk frac_slicers rrz"ArrayOverlapLayer._construct_graphsUyyyyOO% !DJJ. !DJJ. ..LML 5CV$%'' &T  wK 0#g, d A)4'A+t: r!F)r(r)r*r+r r~propertyrr@rrrrrrr_r`s@rrtrtsI ! 7!! -&+r!rtc tdfd}g}t|ddD]Z\}}|j|d}d} |dkDr |ddk7r| dz } ||dz kr |ddk7r| dz } |j| \d} t|ddDcgc]:\}}| |j|dr||||j|dn|g<} }}||gg| z} tt | } t|D cgc]!\}} | |j|dr| nd#}}} t || }|Scc}}wcc} }w)aGet all neighboring keys around center Parameters ---------- k: Key The key around which to generate new keys dims: Sequence[int] The number of chunks in each dimension name: Option[str] The name to include in the output keys, or none to include no name axes: Dict[int, int] The axes active in the expansion. We don't expand on non-active axes Examples -------- >>> _expand_keys_around_center(('x', 2, 3), dims=[5, 5], name='y', axes={0: 1, 1: 1}) # noqa: E501 # doctest: +NORMALIZE_WHITESPACE [[('y', 1.1, 2.1), ('y', 1.1, 3), ('y', 1.1, 3.9)], [('y', 2, 2.1), ('y', 2, 3), ('y', 2, 3.9)], [('y', 2.9, 2.1), ('y', 2.9, 3), ('y', 2.9, 3.9)]] >>> _expand_keys_around_center(('x', 0, 4), dims=[5, 5], name='y', axes={0: 1, 1: 1}) # noqa: E501 # doctest: +NORMALIZE_WHITESPACE [[('y', 0, 3.1), ('y', 0, 4)], [('y', 0.9, 3.1), ('y', 0.9, 4)]] c.t|ts||f}|Sr)rdr;depths r convert_depthz1_expand_keys_around_center..convert_depths%'ENE r!c|}g}|dz dkDr|ddk7r|j|dz |j||dz|dz kr|ddk7r|j|dz|S)Ng?rrV)append)rHrrrvrrs rindsz(_expand_keys_around_center..indssue$  9q=U1X] IIcCi  # 9tAw{ "uQx1} IIcCi  r!rVNrcPt|trtd|DS|dk7S)Nc3&K|] }|dk7 yw)rNr,)r8xs rr:zC_expand_keys_around_center.._valid_depth..0s-u!qAvusr)rdr;anyrs r _valid_depthz0_expand_keys_around_center.._valid_depth.s' eU #-u-- -A: r!) enumerategetrrr reshapelist)rrrvrwrshaperHrrnumrr%seqdshape2rrs ` @rrrsd4  EAabE"3dhhq!n- 7uQx1} 1HC a1 qQ 1HC S# !"& &FAs)5TXXa^(DQTXXa^$3%O&   x$ w~ CCLUCS TCS41a<A/aQ6CSF T  %F M Us ?D.7&D4ct|dk(r t|Stt||dz }tj||Dcgc]}t |dd|c}Scc}w)zgReshape iterator to nested shape >>> reshapelist((2, 3), range(6)) [[0, 1, 2], [3, 4, 5]] rVrN)r7rintr partitionr)rrnparts rrr@sb  5zQCy C58# $9>C9PQ9P E!"It,9PQQQs A%c$|dftd|ddDz}g}tt|dd|ddD]\}\}}|j|d}t |tr |d}|d} n|}|} ||k(r|j t ddd\||kr| r|j t d| ||kDr|r|j t | dyt|}td|Dr|Stj||fS)a >>> fractional_slice(('x', 5.1), {0: 2}) (, ('x', 5), (slice(-2, None, None),)) >>> fractional_slice(('x', 3, 5.1), {0: 2, 1: 3}) (, ('x', 3, 5), (slice(None, None, None), slice(-3, None, None))) >>> fractional_slice(('x', 2.9, 5.1), {0: 2, 1: 3}) (, ('x', 3, 5), (slice(0, 2, None), slice(-3, None, None))) rc3DK|]}tt|ywr)rround)r8rHs rr:z#fractional_slice..Ys A1U1Xs rVNFc3<K|]}|tdddk(ywrrZ)r8rs rr:z#fractional_slice..os ;Uc3%dD) )UrX) r;rrIrrdrr[alloperatorgetitem) taskrwroundedindexrHtrr left_depth right_depths rrrMsAwj5 AQR AAAG Es48WQR[9: 6AqA eU #qJ(KJK 6 LLtT40 1 U{ LLq+. / Uz LL {D1 2!;" %LE ;U ;;   '511r!ceZdZdZ dfd ZdZdZdZdZe dZ dZ d Z d Z d Zdd Zd ZdZddZxZS)SimpleShuffleLayeraSimple HighLevelGraph Shuffle layer High-level graph layer for a simple shuffle operation in which each output partition depends on all input partitions. Parameters ---------- name : str Name of new shuffled output collection. column : str or list of str Column(s) to be used to map rows to output partitions (by hashing). npartitions : int Number of output partitions. npartitions_input : int Number of partitions in the original (un-shuffled) DataFrame. ignore_index: bool, default False Ignore index during shuffle. If ``True``, performance may improve, but index values will not be preserved. name_input : str Name of input collection. meta_input : pd.DataFrame-like object Empty metadata of input collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations c ||_||_||_||_||_||_||_|xs t||_d|jz|_ | xsi} d|_ d| vr|j| d<t |5| y)Nsplit-priorityr)rvcolumn npartitionsnpartitions_input ignore_index name_input meta_inputr parts_out split_name _split_keys _key_priorityrRr ) rrvrrrrrrrrrSs rr zSimpleShuffleLayer.__init__s  &!2($$"8eK&8"TYY."'R  [ (&*&8&8K # [1r!cNt|tsJ|d|jk(ryy)NrrV)rdr;rrs rrz SimpleShuffleLayer._key_prioritys(#u%%% q6T__ $r!cX|jDchc]}|j|fc}Scc}wrrrvrrs rrz"SimpleShuffleLayer.get_output_keys'.2nn=ndD!n==='cNdj|j|jS)Nz-SimpleShuffleLayer)formatrvrr}s rr~zSimpleShuffleLayer.__repr__s$>EE IIt''  r!ct|dSrrr}s rrz"SimpleShuffleLayer.is_materializedrr!cxt|dr |jS|j}||_|jSrrrs rrzSimpleShuffleLayer._dictrr!c |j|Srrrs rr@zSimpleShuffleLayer.__getitem__rr!c,t|jSrrr}s rrzSimpleShuffleLayer.__iter__rr!c,t|jSrrr}s rrzSimpleShuffleLayer.__len__rr!ct}|D])} |\}}||jk7r|j|+|S#t$rY9wxYwz4Simple utility to convert keys to partition indices.set ValueErrorrvaddrrpartsr_name_parts r_keys_to_partsz!SimpleShuffleLayer._keys_to_partsYC " u ! IIe    ; AAc tt}|xs|j|}|D]H}||j|fxxt |j Dchc]}|j |fc}zcc<J|Scc}w)zDetermine the necessary dependencies to produce `keys`. For a simple shuffle, output partitions always depend on all input partitions. This method does not require graph materialization. )rrrrvrrr)rrrdepsrrHs r_cull_dependenciesz%SimpleShuffleLayer._cull_dependenciess~3:!4!4T!: D $))T" #.3D4J4J.K(.K!$.K(  # (sA4 c t|j|j|j|j|j |j |j|SNr)rrvrrrrrrrrs r_cullzSimpleShuffleLayer._cullsI! II KK     " "    OO OO  r!c|j|}|j||}|t|jk7r|j |}||fS||fS)aCull a SimpleShuffleLayer HighLevelGraph layer. The underlying graph will only include the necessary tasks to produce the keys (indices) included in `parts_out`. Therefore, "culling" the layer only requires us to reset this parameter. rrrrrrrrall_keysr culled_deps culled_layers rcullzSimpleShuffleLayer.cull^''- --di-H DNN+ +::i0L, ,$ $r!c @d|jz}|rtd}td}n ddlm}ddlm}i}|j D]}t|jDcgc]}|j||f}}|||jf||j|f<|D]\} } } tj|| f| f||j| | f<|| f|vs3||j| f|jd|j|j|j|jf||| f<|Scc}w)z/Construct graph for a simple shuffle operation.group-dask.dataframe.core._concat$dask.dataframe.shuffle.shuffle_groupr_concat shuffle_group)rvrdask.dataframe.corer&dask.dataframe.shuffler(rrrrrrrrrr) rrshuffle_group_name concat_funcshuffle_group_funcrpart_outpart_in _concat_list_ _part_out_part_ins rrz#SimpleShuffleLayer._construct_graph sS& 1 --JKK!36"  C RH %T%;%;< -!!     % .r!rcDeZdZdZ dfd ZdZddZdZd dZxZ S) ShuffleLayera"Shuffle-stage HighLevelGraph layer High-level graph layer corresponding to a single stage of a multi-stage inter-partition shuffle operation. Stage: (shuffle-group) -> (shuffle-split) -> (shuffle-join) Parameters ---------- name : str Name of new (partially) shuffled collection. column : str or list of str Column(s) to be used to map rows to output partitions (by hashing). inputs : list of tuples Each tuple dictates the data movement for a specific partition. stage : int Index of the current shuffle stage. npartitions : int Number of output partitions for the full (multi-stage) shuffle. npartitions_input : int Number of partitions in the original (un-shuffled) DataFrame. k : int A partition is split into this many groups during each stage. ignore_index: bool, default False Ignore index during shuffle. If ``True``, performance may improve, but index values will not be preserved. name_input : str Name of input collection. meta_input : pd.DataFrame-like object Empty metadata of input collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations c ||_||_||_t ||||||| | | xst t ||  y)N)rr)inputsstagensplitsrRr rr7)rrvrr8r9rrr:rrrrrrSs rr zShuffleLayer.__init__vsU           55V#5#  r!czdj|j|j|j|jS)Nz=ShuffleLayer)rrvr9r:rr}s rr~zShuffleLayer.__repr__s0NUU IItzz4<<1A1A  r!c8tt}|xs|j|}t|jDcic]\}}|| }}}|D]}|j|}t |j D]} t||j| } || } |jdk(r@| |jk\r1||j|fjd|jz| dfn||j|fj|j| f|Scc}}w)zqDetermine the necessary dependencies to produce `keys`. Does not require graph materialization. rr"empty) rrrrr8rr:rr9rrvr r) rrrrrHinp inp_part_maproutr_inpr s rrzShuffleLayer._cull_dependenciess 3:!4!4T!: -6t{{-CD-C61cQ-C DD++d#C4<<(c4::q1$T*::?u0F0F'F$))T*+//DII1EtW0UV$))T*+//%0HI ) Es Dc t|j|j|j|j|j |j |j|j|j|j| Sr) r6rvrr8r9rrr:rrrrs rrzShuffleLayer._culls^ II KK KK JJ     " " LL    OO OO  r!c d|jz}|rtd}td}n ddlm}ddlm}i}t |jDcic]\}}|| }}}|jD]g} |j| } g} t|jD]F}t| |j|} | |j} | j|j| | fH|| |jf||j| f<| D]\}} } t j"|| f| f||j| | f<|| f|vs3|| }|jdk(r3||j$kr|j&|f}n#|| df}|j(||<n|j&|f}|||j*|j|j|j$|j|j,f||| f<j|Scc}}w)z2Construct graph for a "rearrange-by-column" stage.r"r#r$rr%r'r=)rvrr)r&r*r(rr8rrr:rr9rrrrrrrrrr)rrr+r,r-rrHr>r?rr@r0rA_idxr1r  input_keys rrzShuffleLayer._construct_graphs& 1 --JKK!36"  C R-6t{{-CD-C61cQ-C DNND++d#CL4<<(c4::q14::##T__dD$AB )!!&CD! " ". 4$$'.6T__dD12 '-S8(.EzzQ 4#9#99)-%(@I*)rrvrIrLrNr}s rr~zBroadcastJoinLayer.__repr__>s.FMM IItxx   r!ct|dSrrr}s rrz"BroadcastJoinLayer.is_materializedCrr!cxt|dr |jS|j}||_|jSrrrs rrzBroadcastJoinLayer._dictFrr!c |j|Srrrs rr@zBroadcastJoinLayer.__getitem__Prr!c,t|jSrrr}s rrzBroadcastJoinLayer.__iter__Srr!c,t|jSrrr}s rrzBroadcastJoinLayer.__len__Vrr!ct}|D])} |\}}||jk7r|j|+|S#t$rY9wxYwrrr s rrz!BroadcastJoinLayer._keys_to_partsYrrc|j|jkr.|j|j|j|jfS|j|j|j|j fSr)rMrOrLrNrKrJr}s r_broadcast_planz"BroadcastJoinLayer._broadcast_planfsl   $"6"6 6 $$    $$   r!c (|jdd\}}}tt}|xs|j|}|D]P}||j|fxxt |Dchc]}||fc}zcc<||j|fxx||fhzcc<R|Scc}w)zDetermine the necessary dependencies to produce `keys`. For a broadcast join, output partitions always depend on all partitions of the broadcasted collection, but only one partition of the "other" collection. N)rZrrrrvr) rrr bcast_name bcast_size other_namerrrHs rrz%BroadcastJoinLayer._cull_dependenciess.2-A-A"1-E* J 3:!4!4T!: D $))T" #zAR'SARAQAR'S S # $))T" #T"(  #  (Ts B ct|j|j|j|j|j |j f|j|d|jS)N)rr) rGrvrrLrMrNrOrrPrs rrzBroadcastJoinLayer._cullsb! II    MM  MM   ((    r!c|j|}|j||}|t|jk7r|j |}||fS||fS)aCull a BroadcastJoinLayer HighLevelGraph layer. The underlying graph will only include the necessary tasks to produce the keys (indices) included in `parts_out`. Therefore, "culling" the layer only requires us to reset this parameter. rrrs rrzBroadcastJoinLayer.cullr r!cd|jz}d|jz}|r"td}td}td}nddlm}ddlm}dd lm}|j \}}} } |j|jkrd nd } i} |jD]} |jd k7r || | f| |f| || f<g}t|D]p}|jd k7rtj|| f|fn| | f||fg}| d k(r|j|| |f}t|||j f| |<|j#|r||f| |j| f<| S) z/Construct graph for a broadcast join operation.zinter-rz%dask.dataframe.multi._split_partitionz$dask.dataframe.multi._concat_wrapperz)dask.dataframe.multi._merge_chunk_wrapperr)_concat_wrapper)_merge_chunk_wrapper)_split_partitionleftrightinner)rvrdask.dataframe.multircrdrerZrMrOrrIrrrreverserrPr)rr inter_namersplit_partition_funcr,merge_chunk_funcr]r^r_other_on bcast_siderrHr0j _merge_args inter_keys rrz#BroadcastJoinLayer._construct_graphs )  ) $67$ --STK1;   L U U8<7K7K4 J H#33d6J6JJVPW Axx7"(O (ZO$L:& 88w. %,,'O )!_O  ' '')'A. $%% "I ##I.3'8$/ "=CA ] ` r!)NNNNrr)r(r)r*r+r rr~rrrr@rrrrZrrrrr_r`s@rrGrGs|D6:> -!!  2&  % Or!rGcHeZdZdZ dfd ZedZdZdZxZ S)DataFrameIOLayera DataFrame-based Blockwise Layer with IO Parameters ---------- name : str Name to use for the constructed layer. columns : str, list or None Field name(s) to read in as columns in the output. inputs : list or BlockwiseDep List of arguments to be passed to ``io_func`` so that the materialized task to produce partition ``i`` will be: ``(, inputs[i])``. Note that each element of ``inputs`` is typically a tuple of arguments. io_func : callable A callable function that takes in a single tuple of arguments, and outputs a DataFrame partition. Column projection will be supported for functions that satisfy the ``DataFrameIOFunction`` protocol. label : str (optional) String to use as a prefix in the place-holder collection name. If nothing is specified (default), "subset-" will be used. produces_tasks : bool (optional) Whether one or more elements of `inputs` is expected to contain a nested task. This argument in only used for serialization purposes, and will be deprecated in the future. Default is False. creation_info: dict (optional) Dictionary containing the callable function ('func'), positional arguments ('args'), and key-word arguments ('kwargs') used to produce the dask collection with this underlying ``DataFrameIOLayer``. annotations: dict (optional) Layer annotations to pass through to Blockwise. c ||_||_||_||_||_||_||_||_t|ts@tt|jD cic] \} } | f|  c} } |j } n|} t|j|ttd} t |A|jd| | dfgi|ycc} } w)N)r4rrH)outputoutput_indicesrindicesr2r)rv_columnsr8io_funclabelr4r creation_infordrrrr r rrRr )rrvcolumnsr8rzr{r4r|rrHr> io_arg_maprrSs rr zDataFrameIOLayer.__init__'s     ,&*&,/))24;;)?@)?vq#!s)?@#22J  JDIIw0B(CD 99 #&'#  As'C c|jS)z(Current column projection for this layer)ryr}s rr}zDataFrameIOLayer.columnsOs}}r!c ddlm}t|}|j$t |jj |rt |j|r|jj|}n |j}t|jxsddzt|j|z||j||j|j|j}|S|S)zProduce a column projection for this IO layer. Given a list of required output columns, this method returns the projected layer. r)DataFrameIOFunctionsubset-)r{r4r)dask.dataframe.io.utilsrrr}r issupersetrdrzproject_columnsrtr{r rvr8r4r)rr}rrzlayers rrz DataFrameIOLayer.project_columnsTs @w- << 3t||#4#?#?#H$,,(;<,,66w?,,$'x3.$))W1MM jj#22 ,,ELKr!cvdj|jt|j|jS)Nz3DataFrameIOLayer)rrvr7r8r}r}s rr~zDataFrameIOLayer.__repr__ss-DKK IIs4;;'  r!)NFNN) r(r)r*r+r rr}rr~r_r`s@rrtrts;"T& P> r!rtc<eZdZUdZded<ded<ded<ded<ded <d ed <ded <ded <ded<ded<ded<ded< d$ d%fd ZdddZd&dZdZdZ dZ dZ dZ e dZdZdZd Zd!Zd"Zd#ZxZS)'DataFrameTreeReductionagDataFrame Tree-Reduction Layer Parameters ---------- name : str Name to use for the constructed layer. name_input : str Name of the input layer that is being reduced. npartitions_input : str Number of partitions in the input layer. concat_func : callable Function used by each tree node to reduce a list of inputs into a single output value. This function must accept only a list as its first positional argument. tree_node_func : callable Function used on the output of ``concat_func`` in each tree node. This function must accept the output of ``concat_func`` as its first positional argument. finalize_func : callable, optional Function used in place of ``tree_node_func`` on the final tree node(s) to produce the final output for each split. By default, ``tree_node_func`` will be used. split_every : int, optional This argument specifies the maximum number of input nodes to be handled by any one task in the tree. Defaults to 32. split_out : int, optional This argument specifies the number of output nodes in the reduction tree. If ``split_out`` is set to an integer >=1, the input tasks must contain data that can be indexed by a ``getitem`` operation with a key in the range ``[0, split_out)``. output_partitions : list, optional List of required output partitions. This parameter is used internally by Dask for high-level culling. tree_node_name : str, optional Name to use for intermediate tree-node tasks. strrvrrrrr,tree_node_funcCallable | None finalize_func split_every split_outz list[int]output_partitionstree_node_namewidthsheightc t || ||_||_||_||_||_||_||_||_ | "tt|jxsdn| |_ | xsd|jz|_ |j} | g|_| dkDrLtj | |jz } |jj#t%| | dkDrLt'|j|_y)NrrVz tree_node-)rRr rvrrr,rrrrrrrrrmathceilrrr7r)rrvrrr,rrrrrrrr rSs rr zDataFrameTreeReduction.__init__s [1 $!2&,*&"!( t~~*+ ,"  -H tyy0H&&g aiIIed&6&667E KK  s5z *ai$++& r!rsplitc*|jr||fzS|Sr)r)rr name_partss r _make_keyz DataFrameTreeReduction._make_keys)-zUH$FJFr!c|r|jr |j}n |j}tj||j|fSr)rrrrr,)r input_keys final_task outer_funcs r _define_taskz#DataFrameTreeReduction._define_tasks< $,,++J,,J J(8(8*EEr!c i}|js|S|j}|jrd|dz }|jD]P}t|jD]6}t j |j|f|f||j|||<8R|jdk\rm|jD][}td|jD]>}t|j|D]}|j|dz }|j|z}t||jz|} |dk(r,t|| Dcgc]}|j|||} }n9t|| Dcgc]$}|j|j||dz |&} }||jdz k(r3|dk(s Jd|d|j| d ||j|f<|j| d ||j|j|||<"A^|S|jD]9}|j|d|g} |j| d ||j|f<;|Scc}wcc}w) z%Construct graph for a tree reduction.z-splitrrVrzgroup = z%, not 0 for final tree reduction taskT)rF)rrrrrrrrrrrminrrrv) rrname_input_user\prgroupp_maxlstartlstoprs rrz'DataFrameTreeReduction._construct_graphs%%J >> h &N++t556A ((!,GC~qBC7, ;;! ++"1dkk2E!&t{{5'9!: $ EAI 6!%!1!1E!9 #FT-=-=$=u E A:*/vu)=*)=A!%~q J)='**/vu)= **>A!%$($7$7EAIQ!/!"*> '*!DKK!O3!& W!)%0UVW *262C2C *t3D3CA/!% 1 1* 1 O  $$($7$7Q!/!"?";3,\ ++"nn^QanHI &*&7&7 t&7&TTYYN#, G* *s 0I )I cddj|j|j|jS)Nz>DataFrameTreeReduction)rrvrrr}s rr~zDataFrameTreeReduction.__repr__+s(OVV IIt  r!cX|jDchc]}|j|fc}Scc}wr)rrv)rr\s r _output_keysz#DataFrameTreeReduction._output_keys0s*(,(>(>?(>1A(>???rcxt|dr |jS|j}||_|jS)N_cached_output_keys)rrr)r output_keyss rrz&DataFrameTreeReduction.get_output_keys3s; 4. /++ +++-K'2D $'''r!ct|dSrrr}s rrz&DataFrameTreeReduction.is_materialized;rr!cxt|dr |jS|j}||_|jSrrrs rrzDataFrameTreeReduction._dict>rr!c |j|Srrrs rr@z"DataFrameTreeReduction.__getitem__Hrr!c,t|jSrrr}s rrzDataFrameTreeReduction.__iter__Krr!ct|jddxsd|jxsdz}|jr%||jt |j zzS|Sr)sumrrrr7r)r tree_sizes rrzDataFrameTreeReduction.__len__NsWQR).Q4>>3FQG >>t55Drsd" #$%)UU%)== @ @.K K&E*E 3% 3' & -(')'/*)+).)4*4yyxBJ R%2^RRjo%od||~t yt nEUEr!