U /e@sddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl ZddlmZddlmZdd lmZmZmZmZdd lmZmZdd lmZdd lmZmZm Z m!Z!m"Z"m#Z#Gd ddZ$GdddeZ%Gddde%Z&Gddde%Z'GdddeZ(d'ddZ)ddZ*ddZ+GdddeZ,Gdd d e,Z-Gd!d"d"eZ.Gd#d$d$eZ/Gd%d&d&eZ0dS)( annotationsN) defaultdict)Callable)product)Any)map)tokenize) Blockwise BlockwiseDepBlockwiseDepDictblockwise_token)flatten keys_in_tasks)Layer)apply cached_cumsumconcreteinsert stringifystringify_collection_keysc@s eZdZdZddZddZdS)CallableLazyImportzFunction Wrapper for Lazy Importing. This Class should only be used when materializing a graph on a distributed scheduler. cCs ||_dSN) function_path)selfrr//tmp/pip-unpacked-wheel-dbjnr7gq/dask/layers.py__init__)szCallableLazyImport.__init__cOsddlm}||j||S)Nr) import_term)Zdistributed.utilsrr)rargskwargsrrrr__call__,s zCallableLazyImport.__call__N)__name__ __module__ __qualname____doc__rr!rrrrr"src@sfeZdZUdZded<ded<dZded<dd d d Zdd d dZdddddZe ddZ dS)ArrayBlockwiseDepzg Blockwise dep for array-likes, which only needs chunking information to compute its data. tuple[tuple[int, ...], ...]chunkstuple[int, ...] numblocksFboolproduces_tasksr(cCs$||_tdd|D|_d|_dS)Ncss|]}t|VqdSr)len).0chunkrrr Esz-ArrayBlockwiseDep.__init__..F)r(tupler*r,rr(rrrrCszArrayBlockwiseDep.__init__idxcCs tddS)Nz%Subclasses must implement __getitem__)NotImplementedErrorrr5rrr __getitem__HszArrayBlockwiseDep.__getitem__Nzlist[tuple[int, ...]] | None)required_indicescCs d|jiS)Nr(r-)rr9rrr__dask_distributed_pack__Ksz+ArrayBlockwiseDep.__dask_distributed_pack__cCs |f|Srrclsstaterrr__dask_distributed_unpack__Psz-ArrayBlockwiseDep.__dask_distributed_unpack__)N) r"r#r$r%__annotations__r,rr8r: classmethodr>rrrrr&9s  r&c@seZdZdZddddZdS)ArrayChunkShapeDepz(Produce chunk shapes given a chunk indexr)r4cCstddt||jDS)Ncss|]\}}||VqdSrr)r/ir0rrrr1Ysz1ArrayChunkShapeDep.__getitem__..)r2zipr(r7rrrr8XszArrayChunkShapeDep.__getitem__N)r"r#r$r%r8rrrrrAUsrAcs>eZdZUdZded<ddfdd Zddd d ZZS) ArraySliceDepz>Produce slice(s) into the full-sized array given a chunk indexr'startsr-cs$t|tdd|D|_dS)Ncss|]}t|ddVqdS)T)Z initial_zeroN)r)r/crrrr1csz)ArraySliceDep.__init__..)superrr2rEr3 __class__rrras zArraySliceDep.__init__r2r4cCs,tddt||jD}tdd|DS)Ncss&|]\}}||||dfVqdSNr)r/rBstartrrrr1fsz,ArraySliceDep.__getitem__..css|]}t|dVqdS)N)Nslicer/srrrr1gs)r2rCrE)rr5locrrrr8eszArraySliceDep.__getitem__)r"r#r$r%r?rr8 __classcell__rrrHrrD\s rDcszeZdZdZfddZddZeddZdd Zd d Z d d Z ddZ ddZ ddZ dddZeddZZS)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}`` cs2t||_||_||_||_||_d|_dSr)rGrnameaxesr(r*token _cached_keys)rrTrUr(r*rVrHrrrys zArrayOverlapLayer.__init__cCsd|jdS)NzArrayOverlapLayersz>ArrayOverlapLayer._dask_keys..keys..csg|]}|fqSrrrr)rrprrrss)r.range)rindresultr(rprTr*)rrrpsz*ArrayOverlapLayer._dask_keys..keys)rWrTr(r*)rrvrrwr _dask_keyss   zArrayOverlapLayer._dask_keysFcCs |j}|j}|j}|}d|j}d|j}|rsz-ArrayOverlapLayer.__dask_distributed_unpack__)F)r"r#r$r%rr[propertyrcr8rirkrnrqrxr`r@r>rRrrrHrrSjs   +rSc sfddg}t|ddD]>\}}d}|dkr<|d7}||dkrT|d7}||q fddt|ddD}|dk r|gg|}tt|} fddt|D} t| | } | S) aGet all neighboring keys around center Parameters ---------- k: tuple They 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)]] csNg}|ddkr||d|||d|dkrJ||d|S)Ng?rrK)append)rBrurv)rzrrindss  z(_expand_keys_around_center..indsrKNrcs2g|]*\}}t|dfr(||n|gqS)ranyget)r/rBru)rUrrrrssz._expand_keys_around_center..cs*g|]"\}}t|dfr"|ndqS)rrKr)r/rBd)rUrrrss) enumeraterr{r reshapelist) rrzrTrUshaperBrunumrseqZshape2rvr)rUrzrrr~s$      r~csFtdkrt|Stt|d}fddt||DSdS)zgReshape iterator to nested shape >>> reshapelist((2, 3), range(6)) [[0, 1, 2], [3, 4, 5]] rKrcsg|]}tdd|qSrJ)rr/partrrrrs)szreshapelist..N)r.r{intr partition)rrnrrrrs rc Cs|dftdd|ddD}g}tt|dd|ddD]\}\}}||d}t|trz|d}|d} n|}|} ||kr|tdddqF||kr| r|td| qF||kr|r|t| dqF|tddqFt|}tdd|Dr|Stj ||fSdS)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))) rcss|]}tt|VqdSr)rroundrrrrrr18sz#fractional_slice..rKNcss|]}|tdddkVqdSrrM)r/rurrrr1Ns) r2rrCr isinstancerrNalloperatorgetitem) ZtaskrUZroundedindexrBtrdepthZ left_depthZ right_depthrrrr,s( $*     rcseZdZdZd&fdd ZddZddZd d Zd d Ze d dZ ddZ ddZ ddZ ddZd'ddZddZddZddZdd Zed!d"Zd(d$d%ZZS))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 Nc stj| d||_||_||_||_||_||_||_|pBt ||_ d|j|_ |j pZi|_ d|j krri|j d<d|j dd<|j d dd|DdS)Nrsplit-priorityZ__expanded_annotations__cSsi|] }|dqS)rKr)r/_keyrrr sz/SimpleShuffleLayer.__init__..)rGrrTcolumn npartitionsnpartitions_input ignore_index name_input meta_inputrt parts_out split_namerupdateget_split_keys) rrTrrrrrrrrrHrrrxs     zSimpleShuffleLayer.__init__csfddtjDS)Ncs(g|] }jD]}tj||fqqSr)rrr)r/part_inpart_outrZrrrssz5SimpleShuffleLayer.get_split_keys..)rtrrZrrZrrs z!SimpleShuffleLayer.get_split_keyscsfddjDS)Ncsh|]}j|fqSrrYrrZrr sz5SimpleShuffleLayer.get_output_keys..rrZrrZrrqsz"SimpleShuffleLayer.get_output_keyscCsd|j|jS)Nz-SimpleShuffleLayer)formatrTrrZrrrr[szSimpleShuffleLayer.__repr__cCs t|dSrlrmrZrrrrnsz"SimpleShuffleLayer.is_materializedcCs$t|dr|jS|}||_|jSr\r^rarrrrcs  zSimpleShuffleLayer._dictcCs |j|Srrdrerrrr8szSimpleShuffleLayer.__getitem__cCs t|jSrrgrZrrrriszSimpleShuffleLayer.__iter__cCs t|jSrrjrZrrrrkszSimpleShuffleLayer.__len__c CsPt}|D]@}z |\}}Wntk r2Yq YnX||jkr@q ||q |Sz4Simple utility to convert keys to partition indices.set ValueErrorrTaddrrppartsrf_name_partrrr_keys_to_partss    z!SimpleShuffleLayer._keys_to_partscsNtt}|p|}|D].}|j|ffddtjDO<q|S)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. csh|]}j|fqSrrrrrZrrrsz8SimpleShuffleLayer._cull_dependencies..)rrrrTrtr)rrprdepsrrrZr_cull_dependenciess z%SimpleShuffleLayer._cull_dependenciesc Cs&t|j|j|j|j|j|j|j|dSNr)rrTrrrrrrrrrrr_cullszSimpleShuffleLayer._cullcCsD||}|j||d}|t|jkr8||}||fS||fSdS)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. rNrrrrrrrpall_keysrZ culled_deps culled_layerrrrculls   zSimpleShuffleLayer.cullc s0ddddddddd g }ttfd d |DfS) NrTrrrrrrrrc3s|]}t|VqdSrgetattrr/attrrZrrr1 sz0SimpleShuffleLayer.__reduce__..)rr2rattrsrrZr __reduce__s zSimpleShuffleLayer.__reduce__c Cs:ddlm}|j|j|j|j|j|j||jt |j dS)Nr to_serialize)rTrrrrrrr) distributed.protocol.serializerrTrrrrrrr{r)rZ all_hlg_keysZknown_key_dependenciesclientZ client_keysrrrrr:s z,SimpleShuffleLayer.__dask_distributed_pack__csddlm}t|dtr*t|d|d<d|krBt|d|d<|f|jdd}dd|D}||Bfd d|D}t |||d S) Nr dumps_taskrinputsTrcSsi|]\}}t|t|qSrrrr/rvrrrr-szBSimpleShuffleLayer.__dask_distributed_unpack__..csi|]\}}|t|gqSrrrrorrr3srbr) distributed.workerrrr2r{r`itemsrprvalmap)r<r=rb dependenciesrZ layer_dskrrrorr>s z.SimpleShuffleLayer.__dask_distributed_unpack__Fc sdj}|r td}td}nddlm}ddlm}i}jD]fddtjD}||j f|jf<|D]^\}}} t j || f|f|j || f<|| f|krz|j | fjdjjj jf||| f<qzqB|S) z/Construct graph for a simple shuffle operation.group-dask.dataframe.core._concat$dask.dataframe.shuffle.shuffle_groupr_concat shuffle_groupcsg|]}j|fqSr)r)r/rrrrrrsJsz7SimpleShuffleLayer._construct_graph..)rTrdask.dataframe.corerdask.dataframe.shufflerrrtrrrrrrrr) rrshuffle_group_name concat_funcshuffle_group_funcrb _concat_list_Z _part_outZ_part_inrrrr`7sB       z#SimpleShuffleLayer._construct_graph)NN)N)F)r"r#r$r%rrrqr[rnrrcr8rirkrrrrrr:r@r>r`rRrrrHrr[s,%*    rcsbeZdZdZdfdd ZddZddZd d Zfd d Zdd dZ ddZ dddZ Z 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 Nc s@||_||_||_tj|||||| | | p4tt|| d dS)N)rr)rstagensplitsrGrrtr.) rrTrrrrrrrrrrrrHrrrszShuffleLayer.__init__c CsTg}|jD]D}|j|}t|jD]*}|t|j||jt||j|fq"q |Sr) rrrtrrrrrr)rrproutrBrrrrs    zShuffleLayer.get_split_keyscCsd|j|j|j|jS)Nz=ShuffleLayer)rrTrrrrZrrrr[s zShuffleLayer.__repr__c s6ddddddddd d d d g }ttfd d|DfS)NrTrrrrrrrrrrrc3s|]}t|VqdSrrrrZrrr1sz*ShuffleLayer.__reduce__..)rr2rrrZrrszShuffleLayer.__reduce__cs0tj||}|j|d<|j|d<|j|d<|S)Nrrr)rGr:rrr)rrr retrHrrr:s    z&ShuffleLayer.__dask_distributed_pack__c Cstt}|p||}ddt|jD}|D]}|j|}t|jD]j}t||j|}||} |jdkr| |j kr||j |f d|j |dfqF||j |f |j | fqFq.|S)zqDetermine the necessary dependencies to produce `keys`. Does not require graph materialization. cSsi|]\}}||qSrrr/rBinprrrrsz3ShuffleLayer._cull_dependencies..rrempty) rrrrrrtrrrrrTrr) rrprr inp_part_maprrr_inprrrrrs "zShuffleLayer._cull_dependenciesc Cs2t|j|j|j|j|j|j|j|j|j |j |d Sr) rrTrrrrrrrrrrrrrrszShuffleLayer._cullFc Csld|j}|r td}td}nddlm}ddlm}i}ddt|jD}|jD]}|j|}g} t |j D].} t ||j | } ||j } | |j| | fqt|| |jf||j|f<| D]\} } } tj|| f| f||j| | f<|| f|kr|| }|j dkr2||jkr|j|f}n|| d f}|j||<n |j|f}|||j|j |j |j|j|jf||| f<qqV|S) z2Construct graph for a "rearrange-by-column" stage.rrrrrrcSsi|]\}}||qSrrrrrrrsz1ShuffleLayer._construct_graph..r)rTrrrrrrrrrtrrrrrrrrrrrrr)rrrrrrbrrrrrBrZ_idxrrZ input_keyrrrr`sX              zShuffleLayer._construct_graph)NN)N)F) r"r#r$r%rrr[rr:rrr`rRrrrHrrhs0  rcseZdZdZd$fdd ZddZddZd d Zed d Z d dZ ddZ ddZ ddZ eddZddZeddZd%ddZddZdd Zd&d"d#ZZS)'BroadcastJoinLayera;Broadcast-based Join Layer High-level graph layer for a join operation requiring the smaller collection to be broadcasted to every partition of the larger collection. Parameters ---------- name : str Name of new (joined) output collection. lhs_name: string "Left" DataFrame collection to join. lhs_npartitions: int Number of partitions in "left" DataFrame collection. rhs_name: string "Right" DataFrame collection to join. rhs_npartitions: int Number of partitions in "right" DataFrame collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations. **merge_kwargs : **dict Keyword arguments to be passed to chunkwise merge func. Nc  stj|d||_||_||_||_||_||_|pBtt |j|_ | |_ |j d|_ |j d|_|j d|_t|jtrtt|jf|_t|jtrtt|jf|_dS)Nrhowleft_onright_on)rGrrTrlhs_namelhs_npartitionsrhs_namerhs_npartitionsrrtr merge_kwargsrrrrrr{r2) rrTrrrrrrrrrHrrras   zBroadcastJoinLayer.__init__csfddjDS)Ncsh|]}j|fqSrrYrrZrrrsz5BroadcastJoinLayer.get_output_keys..rrZrrZrrq~sz"BroadcastJoinLayer.get_output_keyscCsd|j|j|j|jS)Nz5BroadcastJoinLayer)rrTrrrrZrrrr[s zBroadcastJoinLayer.__repr__cCs t|dSrlrmrZrrrrnsz"BroadcastJoinLayer.is_materializedcCs$t|dr|jS|}||_|jSr\r^rarrrrcs  zBroadcastJoinLayer._dictcCs |j|Srrdrerrrr8szBroadcastJoinLayer.__getitem__cCs t|jSrrgrZrrrriszBroadcastJoinLayer.__iter__cCs t|jSrrjrZrrrrkszBroadcastJoinLayer.__len__c Oslddl}i}|jD]0\}}t|tttfs>||||<q|||<q|j|j |j |j |j |j |j|dS)Nr)rTrrrrrrr)picklerrrstrr{r+dumpsrTrrrrrr)rrr rZ _merge_kwargsrrrrrr:s z,BroadcastJoinLayer.__dask_distributed_pack__cs~ddlm}|di}|||f|jdd}dd|D}||Bfdd|D}t|||d S) NrrrTrcSsi|]\}}t|t|qSrrrrrrrszBBroadcastJoinLayer.__dask_distributed_unpack__..csi|]\}}|t|gqSrrrrorrrsr) rrpoprr`rrprr)r<r=rbrrrrawrrrorr>s   z.BroadcastJoinLayer.__dask_distributed_unpack__c CsPt}|D]@}z |\}}Wntk r2Yq YnX||jkr@q ||q |Srrrrrrrs    z!BroadcastJoinLayer._keys_to_partscCs8|j|jkr |j|j|j|jfS|j|j|j|jfSdSr)rrrrrrrZrrr_broadcast_plans  z"BroadcastJoinLayer._broadcast_plancs||jdd\}}tt}|p(||}|D]H}||j|ffddt|DO<||j|f||fhO<q.|S)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. Ncsh|] }|fqSrrrr bcast_namerrrsz8BroadcastJoinLayer._cull_dependencies..)r rrrrTrt)rrpr bcast_size other_namerrrr rrs( z%BroadcastJoinLayer._cull_dependenciescCs0t|j|j|j|j|j|jf|j|d|jS)N)rr) rrTrrrrrrrrrrrrs zBroadcastJoinLayer._cullcCsD||}|j||d}|t|jkr8||}||fS||fSdS)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. rNrrrrrr s   zBroadcastJoinLayer.cullFcCs2d|j}d|j}|r2td}td}td}n$ddlm}ddlm}dd lm}|j\}}} } |j|jkrtd nd } i} |j D]} |j d kr|| | f| |f| || f<g}t |D]d}|j d krt j || f|fn| | f||fg}| d kr||| |f}t|||jf| |<||q||f| |j| f<q| 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)rTrZdask.dataframe.multirrrr rrrrrtrrreverserrr)rrZ inter_namerZsplit_partition_funcrZmerge_chunk_funcr rrZother_onZ bcast_siderbrBrjZ _merge_argsZ inter_keyrrrr`sZ            z#BroadcastJoinLayer._construct_graph)NN)N)F)r"r#r$r%rrqr[rnrrcr8rirkr:r@r>rr rrrr`rRrrrHrrFs,"      rcs>eZdZdZd fdd ZeddZdd Zd d ZZ 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. NFc s||_||_||_||_||_||_||_||_t|t sZt ddt |jD|jd} n|} |j|t dfi} t j|jd| | dfgi|ddS)NcSsi|]\}}|f|qSrrrrrrrsz-DataFrameIOLayer.__init__..)r,rrB)outputZoutput_indicesrbindicesr*r)rT_columnsrio_funclabelr,r creation_inforr r rr rGr) rrTcolumnsrrrr,rrZ io_arg_maprbrHrrrs.  zDataFrameIOLayer.__init__cCs|jS)z(Current column projection for this layer)rrZrrrrszDataFrameIOLayer.columnsc Csddlm}t|}|jdks.t|j|rt|j|rH|j|}n|j}t |j pXddt |j |||j ||j |j|jd}|S|SdS)zProduce a column projection for this IO layer. Given a list of required output columns, this method returns the projected layer. r)DataFrameIOFunctionNZsubset-)rr,r)Zdask.dataframe.io.utilsr r{rr issupersetrrproject_columnsrrr rTrr,r)rrr rZlayerrrrr#s"   z DataFrameIOLayer.project_columnscCsd|jt|j|jS)Nz3DataFrameIOLayer)rrTr.rrrZrrrr[s zDataFrameIOLayer.__repr__)NFNN) r"r#r$r%rrrr#r[rRrrrHrrms*(  rc s.eZdZUdZded<ded<ded<ded<ded <d ed <ded <ded <ded<ded<ded<ded<d?dddddd dddddd fdd ZddddZd@d!d"Zd#d$Zd%d&Z d'd(Z d)d*Z d+d,Z e d-d.Zd/d0Zd1d2Zd3d4Zd5d6Zd7d8Zd9d:Zd;d<Zed=d>ZZS)ADataFrameTreeReductionagDataFrame 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. rrTrrrrrtree_node_funczCallable | None finalize_func split_every split_outz list[int]output_partitionstree_node_namewidthsheightN z int | Nonezlist[int] | Nonez str | Nonezdict[str, Any] | None) rTrrrr%r&r'r(r)r*rc stj| d||_||_||_||_||_||_||_||_ | dkrXt t |j pRdn| |_ | pjd|j|_ |j} | g|_| dkrt| |j} |jt| q|t|j|_dS)NrrKz tree_node-)rGrrTrrrr%r&r'r(r{rtr)r*r+mathceilrrr.r,) rrTrrrr%r&r'r(r)r*rrrHrrrs(zDataFrameTreeReduction.__init__rsplitcGs|jr||fS|Sr)r()rr1Z name_partsrrr _make_key?sz DataFrameTreeReduction._make_keyFcCs(|r|jr|j}n|j}tj||j|fSr)r&r%rrr)r input_keys final_taskZ outer_funcrrr _define_taskEs z#DataFrameTreeReduction._define_taskc si}js|Sjjr`d7jD]6tjD]&}tjj|ff|j|d<q6q(jdkrpjD]tdjD]tj D]}j d}j |}t |j |}dkrfddt||D}nfddt||D}jdkrF|dks,t d |d j |d d |jf<qj |d d |jj|d<qqqrn8jD]0jddg}j |d d |jf<qv|S)z%Construct graph for a tree reduction.z-splitr0rKcsg|]}j|dqS)r0)r2r/p)name_input_userPrrrrsrsz;DataFrameTreeReduction._construct_graph..cs$g|]}jj|ddqS)rKr0)r2r*r7)rrPrrrrsxsrzgroup = z%, not 0 for final tree reduction taskT)r4F)r)rr(rtrrrr2r,r+r'minAssertionErrorr5rTr*)rrbr8groupZp_maxZlstartZlstopr3r)rr9rPrrr`Ms`        z'DataFrameTreeReduction._construct_graphcCsd|j|j|jS)Nz>DataFrameTreeReduction)rrTrr(rZrrrr[s zDataFrameTreeReduction.__repr__csfddjDS)Ncsh|]}j|fqSrrYrOrZrrrsz6DataFrameTreeReduction._output_keys..)r)rZrrZr _output_keyssz#DataFrameTreeReduction._output_keyscCs$t|dr|jS|}||_|jS)N_cached_output_keys)r_r>r=)rZ output_keysrrrrqs  z&DataFrameTreeReduction.get_output_keyscCs t|dSrlrmrZrrrrnsz&DataFrameTreeReduction.is_materializedcCs$t|dr|jS|}||_|jSr\r^rarrrrcs  zDataFrameTreeReduction._dictcCs |j|Srrdrerrrr8sz"DataFrameTreeReduction.__getitem__cCs t|jSrrgrZrrrriszDataFrameTreeReduction.__iter__cCs>t|jddpd|jpd}|jr:||jt|jS|S)NrK)sumr+r(rr.r))rZ tree_sizerrrrks zDataFrameTreeReduction.__len__c CsPt}|D]@}z |\}}Wntk r2Yq YnX||jkr@q ||q |S)z;Simple utility to convert keys to output partition indices.r)rrpsplitsrfr_splitrrr_keys_to_output_partitionss    z1DataFrameTreeReduction._keys_to_output_partitionsc Cs2t|j|j|j|j|j|j|j|j||j |j d S)N)r&r'r(r)r*r) r$rTrrrr%r&r'r(r*r)rr)rrrrszDataFrameTreeReduction._cullcsXjdffddtjDi}|}|tjkrL|}||fS|fSdS)z2Cull a DataFrameTreeReduction HighLevelGraph layerrcsh|]}j|fqSrrrrrZrrrsz.DataFrameTreeReduction.cull..N)rTrtrrBrr)r)rrprrr)rrrZrrs   zDataFrameTreeReduction.cullc Os^ddlm}||j}||j}|jr2||j}nd}|j|j|j||||j|j |j |j d S)Nrr) rTrrrr%r&r'r(r)r*) rrrr%r&rTrrr'r(r)r*)rrr rZ _concat_funcZ_tree_node_funcZ_finalize_funcrrrr:s"    z0DataFrameTreeReduction.__dask_distributed_pack__csdddlm}|f|}dd|D}||Bfdd|D}t|||dS)NrrcSsi|]\}}t|t|qSrrrrrrr szFDataFrameTreeReduction.__dask_distributed_unpack__..csi|]\}}|t|gqSrrrrorrrsr)rrr`rrprr)r<r=rbrrr rrrorr>s   z2DataFrameTreeReduction.__dask_distributed_unpack__)Nr-NNNN)F)r"r#r$r%r?rr2r5r`r[r=rqrnrrcr8rirkrBrrr:r@r>rRrrrHrr$sL % (' J  r$)NN)1 __future__rr|r.r collectionsrcollections.abcr itertoolsrtypingrZtlzrZ tlz.curriedrZ dask.baser Zdask.blockwiser r r r Z dask.corerrZdask.highlevelgraphrZ dask.utilsrrrrrrrr&rArDrSr~rrrrrrr$rrrrs@         ~ 7 /_)x