0FieBUddlmZddlZddlZddlmZddlmZmZm Z m Z m Z ddl m Z ddlmZddlmZddlmZdd lmZmZddlZdd lmZddlZdd lmZdd lmZdd lm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.ddl/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7ddl8m9Z9ddl:m;Z;ddlm?Z?ej@dZAerddlBZCddlDZEddlFmGZGddlHmIZIdId)ZJdJd,ZKdKd/ZL dLdMd3ZMeNe"eOfZPd4eQd5<Gd6d7e ZRdNd;ZSdOd?ZTGd@dAe2eUdfZVedBCGdDdEe3eUZWdPdHZXdS)Q annotationsN) defaultdict)Callable CollectionIterableIteratorSequence)ThreadPoolExecutor) dataclass)partial)Path) TYPE_CHECKINGAny)IOLoop)tokenize)HighLevelGraph)Layer)Key) PooledRPCCall) Reschedule)check_dtype_supportcheck_minimal_arrow_versionconvert_shardsdeserialize_tablelist_of_buffers_to_tableread_from_diskserialize_table)NDIndex ShuffleId ShuffleRun ShuffleSpec barrier_keyget_worker_pluginhandle_transfer_errorshandle_unpack_errors)ResourceLimiter)ShuffleSchedulerPlugin)ShuffleWorkerPlugin)sizeofzdistributed.shuffle) TypeAlias) DataFrameinput pd.DataFrameidr input_partitionint npartitionscolumnstrmeta parts_outset[int]diskboolreturnc t|5t||t||||||cdddS#1swxYwYdS)N)r/r2r3r5r6r8)spec)r%r$ add_partitionDataFrameShuffleSpec)r-r/r0r2r3r5r6r8s z*rearrange_by_column_p2p..xs/ 8 8az!S!! ! 8 8 8 8 8 8cXi|]'}t|t|t|(S)rUr4typerVs r? z+rearrange_by_column_p2p..ys/RRRaz!S?Q?QRq$q''RRRrZz0p2p requires all column names to be str, found: z shuffle_p2p-zdistributed.p2p.disk)npartitions_input name_input meta_inputr8)pandasdask.dataframe.corerR_metaapitypesis_integer_dtypedtype TypeErrorrr2ranycolumnsdaskconfiggetP2PShuffleLayer_namerfrom_collections) rOr3r2pdrRr5token unsupportednamer8layers r?rearrange_by_column_p2prxfs 111111 8D 6< ( (f); < <  XV X X4)r]__name__rvr2rs r?__repr__zP2PShuffleLayer.__repr__s1Dzz" X X49 X XTEU X X X rZset[Key]c*fdjDS)Nc"h|] }j|f Sr\rv)rWpartrs r? z2P2PShuffleLayer.get_output_keys..s ===dD!===rZr6rs`r?get_output_keyszP2PShuffleLayer.get_output_keyss====dn====rZc"t|dS)N _cached_dict)hasattrrs r?is_materializedzP2PShuffleLayer.is_materializedst^,,,rZrycx|t|dr|jS|}||_|jS)z$Materialize full dict representationr)rr_construct_graph)rdsks r?_dictzP2PShuffleLayer._dictsG + 4 ( ( $$ $''))C #D   rZkeyrtuplec|j|SrD)r)rrs r? __getitem__zP2PShuffleLayer.__getitem__sz#rZ Iterator[Key]c*t|jSrD)iterrrs r?__iter__zP2PShuffleLayer.__iter__sDJrZc*t|jSrD)lenrrs r?__len__zP2PShuffleLayer.__len__s4:rZ Iterable[int]c vt|j|j|j|j|j|j|j|S)Nr)rprvr3r2r_r`rar8)rr6s r?_cullzP2PShuffleLayer._cullsA I K    " O O I    rZkeys Iterable[Key]ct}|D]f}t|trOt|dkr<|\}}||jkr,t|t sJ||g|S)z4Simple utility to convert keys to partition indices.)r~rUrrrvr1add)rrpartsrrvrs r?_keys_to_partszP2PShuffleLayer._keys_to_partss{ $ $C#u%% $#c((a-- d49$$%dC00000IIdOOO rZall_keysCollection[Key]tuple[P2PShuffleLayer, dict]c|}tfdtjDfd|D}|t jkr|}||fS|fS)aCull a P2PShuffleLayer 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. c"h|] }j|f Sr\)r`)rWirs r?rz'P2PShuffleLayer.cull..s I I Iadoq ! I I IrZc$i|] }j|f Sr\r)rWr input_partsrs r?r^z(P2PShuffleLayer.cull..s"LLL$ 4(+LLLrZ)r frozensetrr_r~r6r)rrrr6 culled_deps culled_layerrs` @r?cullzP2PShuffleLayer.culls''--  I I I I51G+H+H I I I  MLLLL)LLL DN++ + +::i00L, ,$ $rZc t|j|j|j|j}i}t t |}d|z}t}t|j D]N}| ||ft|j|f|||j|j|j |j|j f |||f<Ot||f||<|j}|jD]}t |||f|||f<|S)Nzshuffle-transfer-)rr`r3r2r6r#r listrr_appendr@rar8rNrvrF)rrtr _barrier_keyrv transfer_keysrpart_outs r?rz P2PShuffleLayer._construct_graphs $+t7GXX ""9U#3#344 "U* t-..  A  $ + + + !$    Cq NN-e]CLy  H %Cx ! !  rZ)NN)rvr4r3r4r2r1r_r1r`r4rar.r8r9r6r{rr|)r:r4)r:r)r:r9)r:ry)rrr:r)r:r)r:r1)r6rr:rp)rrr:r7)rrrrr:r)r __module__ __qualname____annotations__rrrrpropertyrrrrrrrr __classcell__rs@r?rprps IIIKKKOOOJJJ+/#'22222224    >>>>---- ! ! !X !             %%%%.rZrp worker_for pd.Seriesdict[Any, pa.Table]c ddl}ddlm}|j}t |t sJ||jj d|dd}t|}|siS||d d | d} dg ~||d d|dd kdd z} |dg| g} fd t!jd | D} |  | d d || } fdt)| | D} t+t-t| |ksJ| S)zO Split data into many arrow batches, partitioned by destination worker rN)to_pyarrow_table_dispatch_workerTinner)rightleft_on right_indexhow)preserve_indexrbcJg|]\}}|||z  SoffsetlengthslicerWabts r? z#split_by_worker..N>,0AqqQ''rZrrc<i|]\}}jj||Sr\)cat categories)rWcodeshardrs r?r^z#split_by_worker..Ts;    D% !$'   rZ)numpydask.dataframe.dispatchr_constructor_slicedrUr]mergercodesrenamersort_byasarraydropwhere concatenatetoolzsliding_windowrrzipsummapvalues)rOr3r5rnpr constructornrowsrsplitsshards unique_codesoutrs ` @r?split_by_workerr(s AAAAAA(K k4 ( ((((Z((J n")))44    B GGE   "!"T:::A )A JJq| $ $E  {A XXeABBi5":- . .q 1A 5F ^^aS&M * *F494HF4S4SF MM!''D'99:::=L    |V44   C s3 %% & &% / / / / JrZrpa.Tabledict[int, pa.Table]c2ddl}|g|}||||}||dd|ddkddz}|dg|g}fdtj d|D}| |ddtttt|ksJt|t|ksJt!t#||S)zL Split data into many arrow batches, partitioned by final partition rNrbrcJg|]\}}|||z  Srrrs r?rz&split_by_partition..krrZrr)rselect to_pandasuniquesortrrrrrrrrrrrdictr)rr3r partitions partitionrrs` r?split_by_partitionr ]sy6(##--//7>>@@JOO &A 1V9%%I XXimy"~5 6 6q 9A =F ^^aS&M * *F494HF4S4SF MM!''D'99::: q66SS&))** * * * * z??c&kk ) ) ) ) J'' ( ((rZceZdZUdZded<ded<ded<ded <d5fd Zd6d#Zd7d&Zd8d+Zd9d-Z d:d.Z d;d2Z d worker_address. column: The data column we split the input partition by. id: A unique `ShuffleID` this belongs to. run_id: A unique identifier of the specific execution of the shuffle this belongs to. local_address: The local address this Shuffle can be contacted by using `rpc`. directory: The scratch directory to buffer data in. executor: Thread pool to use for offloading compute. rpc: A callable returning a PooledRPCCall to contact other Shuffle instances. Typically a ConnectionPool. scheduler: A PooledRPCCall to to contact the scheduler. memory_limiter_disk: memory_limiter_comm: A ``ResourceLimiter`` limiting the total amount of memory used in either buffer. r4r3r.r5zdict[str, list[int]] partitions_ofrrdict[int, str]r/r run_idr1 local_address directoryexecutorr rpcCallable[[str], PooledRPCCall] schedulerrmemory_limiter_diskr'memory_limiter_commsr8r9looprc ddl}t|||||| | | | | | ||_||_t t }|D] \}}|||!t||_ |j |d d|_ dS)Nr) r/rrrrrrrrr8r_workersrcategory)rcrrr3r5rritemsrrr Seriesastyper)rrr3r5r/rrrrrrrrr8rrsr raddrrs r?rzDataFrameShuffleRun.__init__s"  '!5 3    #D)) $**,, - -JD$ $  & &t , , , ,!-00#")JZ@@@GG SSrZdatalist[tuple[int, bytes]]r:NonecK|g}|D]i}|d|jvrX||d|j|d|xjt |z c_j~|sdS ||j|d{V}~||d{VdS#t$r }||_ d}~wwxYw)Nrrb) raise_if_closedreceivedrr total_recvdr*offload_repartition_buffers_write_to_diskrK _exception)rrfiltereddgroupsrMs r?_receivezDataFrameShuffleRun._receives  . .At4=((!%%% !!!A$'''  F1II-    F <<(A8LLLLLLLLF%%f-- - - - - - - - - -   DO  s =C C!CC! list[bytes]dict[NDIndex, bytes]ct|}t||j}t|t t t|ksJ~d|DS)Nc6i|]\}}|ft|Sr\r)rWkvs r?r^z.s(DDDTQoa((DDDrZ)rr r3rrrrr)rrtabler,s r?r'z(DataFrameShuffleRun._repartition_bufferssp(..#E4;775zzSS&--//!:!:;;;;;; DDV\\^^DDDDrZ partition_idkwargsrdict[str, tuple[int, bytes]]c t||j|j|j}fd|D}|S)Nc:i|]\}}|t|fSr\r2)rWr3rr6s r?r^z8DataFrameShuffleRun._shard_partition..s,MMMAq<!3!34MMMrZ)rr3r5rr)rrr6r7rs ` r?_shard_partitionz$DataFrameShuffleRun._shard_partitionsO   K I O    NMMMMMM rZrc  ||f}t||jS#t$r|jcYSwxYwrD)_read_from_diskrr5KeyErrorcopy)rr6rr7rs r?_get_output_partitionz)DataFrameShuffleRun._get_output_partitionsa  $''88D!$ 22 2 $ $ $9>>## # # # $s*-#AAc|j|SrD)r)rr/s r?_get_assigned_workerz(DataFrameShuffleRun._get_assigned_workersr""rZpathrtuple[pa.Table, int]c t|SrD)r)rrCs r?readzDataFrameShuffleRun.readsd###rZbufferc t|SrD)r)rrGs r? deserializezDataFrameShuffleRun.deserializes (((rZ)rr r3r4r5r.r/r rr1rr4rr4rr rrrrrr'rr'r8r9rr)rr r:r!)rr.r:r/)rr.r6r1r7rr:r8)r6r1rr4r7rr:r.)r/r1r:r4)rCrr:rD)rGrr:r)rrr__doc__rrr-r'r;r@rBrFrIrrs@r?r r ts$$LKKK''''&T&T&T&T&T&TP(EEEE     $ $ $ $####$$$$))))))))rZr T)frozencHeZdZUded<ded<ded<ded<dd ZddZdS)r>r1r2r4r3r.r5r7r6pluginr(r:r cxtt|j}||j|j|SrD)r _get_worker_for_range_shardingr2_pin_output_workersr/r6)rrM pick_workers r?rPz(DataFrameShuffleSpec._pin_output_workerss1NOO ))$'4>;OOOrZrrr)r!cdt|j|j||j|tj|jjd|jd||j |jj |jj |jj |j r|jntd|j|j |jjS)Nzshuffle--)r3r5rr/rrrrrrrrr8r)r r3r5r/osrCjoinworkerlocal_directory _executoraddressrrr8rr'rr)rrrrMs r?create_run_on_workerz)DataFrameShuffleSpec.create_run_on_workers#;!wgll --47--V--% -/ !m-y!' : : &&!'!<#'    rZN)rMr(r:r )rr1rr rMr)r:r!)rrrrrPrZr\rZr?r>r>soKKKPPPP      rZr>workers Sequence[str]c<t||z|z}||S)zKGet address of target worker for this output partition using range sharding)r)r2rAr[rs r?rOrO-s$ G '';6A 1:rZ)r-r.r/r r0r1r2r1r3r4r5r.r6r7r8r9r:r1)r/r rAr1rBr1r:r.)r/r rGrHr:r1rD)rOr,r3r4r2rPr:r,) rOr.r3r4r5r.rrr:r)rrr3r4r:r)r2r1rAr1r[r\r:r4)Y __future__rloggingrT collectionsrcollections.abcrrrr r concurrent.futuresr dataclassesr functoolsr pathlibrtypingrrrtornado.iolooprrm dask.baserdask.highlevelgraphr dask.layersr dask.typingrdistributed.corerdistributed.exceptionsrdistributed.shuffle._arrowrrrrrrrdistributed.shuffle._corerr r!r"r#r$r%r&distributed.shuffle._limiterr'%distributed.shuffle._scheduler_pluginr("distributed.shuffle._worker_pluginr)distributed.sizeofr* getLoggerloggerrcrspyarrowpatyping_extensionsr+dask.dataframer,r@rFrNrxrrryrrprr r1r r>rOr\rZr?rzs6""""""" ######NNNNNNNNNNNNNN111111!!!!!!%%%%%%%% !!!!!! ......******------                    988888HHHHHHBBBBBB%%%%%%  0 1 1),+++++((((((    2    QQQQ#)))))X#3:.....PPPPPePPPf2222j)))).Q)Q)Q)Q)Q)*S.%89Q)Q)Q)h $! ! ! ! ! ;s+! ! ! HrZ