U /eT@sddlmZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z ddl mZddlmZmZmZmZddlZddlmZddlmZdd lmZdd lmZmZmZmZm Z dd l!m"Z"dd l#m$Z$dd l%m&Z&ddl'm(Z(ddl)m*Z*m+Z+er&ddl,Z-ddl.Z/ddl0m1Z1edZ2e3e4Z5Gddde6Z7GdddZ8GdddZ9dddddddZ:dddd d!d"Z;dS)#) annotationsN) defaultdict)CallableIterator)ThreadPoolExecutor) TYPE_CHECKINGAnyTypeVaroverload) parse_bytes) PooledRPCCall) to_serialize)deserialize_schema dump_shardslist_of_buffers_to_tableload_partitionserialize_table)CommShardsBuffer)DiskShardsBuffer)ResourceLimiter) ShuffleId) log_errorssync)WorkerTc@s eZdZdS)ShuffleClosedErrorN)__name__ __module__ __qualname__rrI/tmp/pip-unpacked-wheel-g426oqom/distributed/shuffle/_worker_extension.pyr,src @s|eZdZdZdddddddddd d d d d d ZddddZejdddddZddddZ ddddddZ ddddd d!Z d"dd#d$Z ddd%d&d'Z ddd%d(d)Zdd*d%d+d,Zd*dd%d-d.Zddd/d0Zd1dd%d2d3Zd4dd%d5d6Zdd1d7d8d9Zd:d;d<d=d>Zddd?d@ZdddAdBZdCddDdEZdddFdGZdddHdIZdJddKdLdMZdNS)OShuffleaoState for a single active shuffle This object is responsible for splitting, sending, receiving and combining data shards. It is entirely agnostic to the distributed system and can perform a shuffle with other `Shuffle` instances using `rpc` and `broadcast`. The user of this needs to guarantee that only `Shuffle`s of the same unique `ShuffleID` interact. Parameters ---------- worker_for: A mapping partition_id -> worker_address. output_workers: A set of all participating worker (addresses). column: The data column we split the input partition by. schema: The schema of the payload data. id: A unique `ShuffleID` this belongs to. local_address: The local address this Shuffle can be contacted by using `rpc`. directory: The scratch directory to buffer data in. nthreads: How many background threads to use for compute. loop: The event loop. rpc: A callable returning a PooledRPCCall to contact other Shuffle instances. Typically a ConnectionPool. broadcast: A function that ensures a RPC is evaluated on all `Shuffle` instances of a given `ShuffleID`. memory_limiter_disk: memory_limiter_comm: A ``ResourceLimiter`` limiting the total amount of memory used in either buffer. zdict[int, str]setstrz pa.SchemarintzCallable[[str], PooledRPCCall]rr) worker_foroutput_workerscolumnschemaid local_address directorynthreadsrpc broadcastmemory_limiter_diskmemory_limiter_commsc Csddl} | |_| |_||_||_||_||_t||_t t }||_ | D]\}}|| |qLt||_| j|ddd|_d|_ttt|| d|_t|j| d|_t t|_t|j|d|_d|_ d|_!t""|_#d|_$t%&|_'dS) NrZ_workers)namecategoryF)dumploadr+memory_limiter)sendr5r)(pandasr.r-r'r)r(r&rexecutorrlistr*itemsappenddict partitions_ofZSeriesZastyper%closedrrr _disk_bufferrr6 _comm_bufferfloat diagnosticslengetoutput_partitions_left transferred total_recvdtime start_time _exceptionasyncioEvent _closed_event)selfr%r&r'r(r)r*r+r,r-r.r/r0pdr=partaddrrrr __init__\s@    zShuffle.__init__returncCsd|jd|jdS)Nz )r)r*rNrrr __repr__szShuffle.__repr__zIterator[None])r1rTccs0t}dVt}|j|||7<dSN)rHrB)rNr1startstoprrr rHsz Shuffle.timeNonecs&||jd|jddIdHdS)Nshuffle_inputs_done)op shuffle_id)msg)raise_if_closedr.r)rWrrr barrierszShuffle.barrier list[bytes])addressshardsrTcs(|||jt||jdIdHS)N)datar_)rar-shuffle_receiver r))rNrdrerrr r6s  z Shuffle.sendzCallable[..., T]rr)funcargsrTc sH||d,tj|j|f|IdHW5QRSQRXdS)Ncpu)rarHrKZget_running_loopZrun_in_executorr8)rNrhrirrr offloads zShuffle.offloadzdict[str, Any]cCs,|j}|j|d<|j||j|jdS)Nread)ZdiskZcommrBrZ)r@ heartbeatrGr?rBrI)rNZcomm_heartbeatrrr rms  zShuffle.heartbeat)rfrTcs||IdHdSrY)_receiverNrfrrr receiveszShuffle.receivec sx|z@|jttt|7_||j|IdH}||IdHWn*tk rr}z ||_ W5d}~XYnXdSrY) rarGsummaprCrk_repartition_buffers_write_to_disk ExceptionrJ)rNrfgroupserrr rnszShuffle._receivezdict[str, list[pa.Table]]cCsFt|}t||j}t|ttt|ks2t~dd|DS)NcSsi|]\}}||gqSrr).0kvrrr sz0Shuffle._repartition_buffers..) rsplit_by_partitionr'rCrqrrvaluesAssertionErrorr:)rNrftablervrrr rss  zShuffle._repartition_bufferscs||j|IdHdSrY)rar?writerorrr rtszShuffle._write_to_diskcCs.|jr*|jr|jtd|jd|jdS)NzShuffle z has been closed on )r>rJrr)r*rWrrr ras zShuffle.raise_if_closed pd.DataFramecsTjrtdddfdd }|IdH}|IdHdS)Nz&Cannot add more partitions to shuffle dict[str, list[bytes]]rScs&tjj}dd|D}|S)NcSsi|]\}}|t|gqSr)r)rxrytrrr r{sz4Shuffle.add_partition.._..)split_by_workerr'r%r:)outrfrNrr _sz Shuffle.add_partition.._)rarF RuntimeErrorrk_write_to_comm)rNrfrrrrr add_partitions  zShuffle.add_partitionrcs||j|IdHdSrY)rar@rrorrr rszShuffle._write_to_comm)irTc s||jstd|j||jksJtd|d|j|d|jd|jdksltd|d|jd |IdHz,||}|d | }W5QRXWn"t k r|j }YnX|jd 8_|S) Nz1`get_output_partition` called before barrier taskzOutput partition z belongs on z, not z. rz5No outputs remaining, but requested output partition rU.rj) rarFr~r%r*rE flush_receive_read_from_diskrH to_pandasKeyErrorr(Z empty_table)rNrdfrrrr get_output_partitions$  zShuffle.get_output_partitionz int | strpa.Table)r)rTcCs||j|SrY)rar?rl)rNr)rrr r szShuffle._read_from_diskc sh||jrtdd|_|IdHz|jWn*tk rb}z ||_W5d}~XYnXdS)Nz#`inputs_done` called multiple timesT)rarFr~ _flush_commr@Zraise_on_exceptionrurJ)rNrwrrr inputs_doneszShuffle.inputs_donecs||jIdHdSrY)rar@flushrWrrr rszShuffle._flush_commboolcCs|jo|jdkS)Nr)rFrErWrrr donesz Shuffle.donecs||jIdHdSrY)rar?rrWrrr r szShuffle.flush_receivecs|jr|jIdHdSd|_|jIdH|jIdHz|jjddWntk rp|jYnX|j dS)NT)Zcancel_futures) r>rMwaitr@closer?r8shutdownrur"rWrrr r$sz Shuffle.closeru) exceptionrTcCs|js ||_dSrY)r>rJ)rNrrrr fail2sz Shuffle.failN)rrr__doc__rRrX contextlibcontextmanagerrHrbr6rkrmrprnrsrtrarrrrrrrrrrrrrr r!0s0+$6    r!c@seZdZUdZded<ded<ded<ded<d ed <dd d d dZddddZddd dddZdd dddZddd dddZ ddd dd d!d"d#Z dd dd$d%Z d&d d'd(d)Z e dd&dd*d+Ze dddd d&d,d-d+ZdAdd/d0d1d&d,d2d+Zdddd3d4d5Zd dd6d7Zdd dd8d9Ze dddd d&d,d:d;Ze dd&ddd?d@Zd.S)CShuffleWorkerExtensionaInterface between a Worker and a Shuffle. This extension is responsible for - Lifecycle of Shuffle instances - ensuring connectivity between remote shuffle instances - ensuring connectivity and integration with the scheduler - routing concurrent calls to the appropriate `Shuffle` based on its `ShuffleID` - collecting instrumentation of ongoing shuffles and route to scheduler/worker rworkerzdict[ShuffleId, Shuffle]shufflesrr0r/rr>r\)rrTcCsl|j|jd<|j|jd<|j|jd<|j|jd<||jd<||_i|_tt d|_ tt d|_ d|_ dS) Nrgr] shuffle_failz shuffle-failshufflez100 MiBz1 GiBF) rghandlersr]rZstream_handlers extensionsrrrr r0r/r>)rNrrrr rRIs     zShuffleWorkerExtension.__init__r<rScCsdd|jDS)NcSsi|]\}}||qSr)rm)rxr)rrrr r{]sz4ShuffleWorkerExtension.heartbeat..)rr:rWrrr rm\sz ShuffleWorkerExtension.heartbeatrrc)r_rfrTcs$||IdH}||IdHdS)z Handler: Receive an incoming shard of data from a peer worker. Using an unknown ``shuffle_id`` is an error. N) _get_shufflerp)rNr_rfrrrr rg_s z&ShuffleWorkerExtension.shuffle_receive)r_rTc spt`||IdH}|IdH|rb|jjs:ttd|| |IdH|j |=W5QRXdS)z Handler: Inform the extension that all input partitions have been handed off to extensions. Using an unknown ``shuffle_id`` is an error. NzShuffle inputs done ) rrrrr?emptyr~loggerinfo_register_completerrNr_rrrr r]ks z*ShuffleWorkerExtension.shuffle_inputs_doner#)r_messagerTcsRz|j|}Wntk r$YdSXt|}|||IdH|j|=dSrY)rrrrr)rNr_rrrrrr r|s z#ShuffleWorkerExtension.shuffle_failrr$)rfr_ npartitionsr'rTcCs*|j||||d}t|jj|j|ddS)N)rrr')rf) get_shufflerrloopr)rNrfr_rr'rrrr rsz$ShuffleWorkerExtension.add_partitioncs"||IdH}|IdHdS)z Task: Note that the barrier task has been reached (`add_partition` called for all input partitions) Using an unknown ``shuffle_id`` is an error. Calling this before all partitions have been added is undefined. N)rrbrrrr _barriers zShuffleWorkerExtension._barrierr!)rrTcs0|IdH|jjj|j|jjdIdHdS)N)r)r)rr schedulerZshuffle_register_completer)rd)rNrrrr rs z)ShuffleWorkerExtension._register_completecsdSrYrrNr_rrr rsz#ShuffleWorkerExtension._get_shuffle)r_rr'rrTcsdSrYrrNr_rr'rrrr rsNzpd.DataFrame | Nonez str | Nonez int | Nonec sddl}z|j|}WnZtk rrzj|jjj||dk rR|j| nd|||jj dIdH}|ddkrt |d|ddkst Wn0tk rt ddd lm}|YnX|jrt|jjd |jj ||jkr`t|d |d |d t|d|tj|jjd||jjj|jj |jjt|j ||j!|j"d }||j|<|j|YSYnX|j#r|j#|SdS)z=Get a shuffle by ID; raise ValueError if it's not registered.rN)r)r(rr'rstatusERRORrOKzEWorker Shuffle unable to get information from scheduler, rescheduling) Reschedulez already closed on r'r%r&r(zshuffle-) r'r%r&r(r)r+r,r*r-r.r/r0)$pyarrowrrrrZ shuffle_getZSchema from_pandas serializeZ to_pybytesrdrr~rrdistributed.workerrr>r __class__rr!rospathjoinZlocal_directorystater,r- functoolspartial_broadcast_to_participantsr/r0rJ) rNr_rr'rparresultrrrr rsj       )r)r`rTcs.|jjj|dIdH}|jjj||dIdHS)N)r))r`workers)rrZ!shuffle_get_participating_workersr.)rNr)r`Zparticipating_workersrrr rs z1ShuffleWorkerExtension._broadcast_to_participantscs8|jr td|_|jr4|j\}}|IdHqdS)NT)r>r~rpopitemr)rNrrrrr rs  zShuffleWorkerExtension.closecCst|jj|j|dSrY)rrrrrrrr rbszShuffleWorkerExtension.barriercCsdSrYrrrrr rsz"ShuffleWorkerExtension.get_shufflecCsdSrYrrrrr rscCst|jj|j||||SrY)rrrrrrrr r#s)r_output_partitionrTcCsP||}t|jj|j|}|rL||jkrL|j|}t|jj|j||S)z Task: Retrieve a shuffled output partition from the ShuffleExtension. Calling this for a ``shuffle_id`` which is unknown or incomplete is an error. ) rrrrrrrpopr)rNr_rroutputrrr r3s   z+ShuffleWorkerExtension.get_output_partition)NNN)NNN)rrrr__annotations__rRrmrgr]rrrrr rrrrbrrrrrr r7sD        C   rrr#z pd.Serieszdict[Any, pa.Table])rr'r%rTc s"ddl}ddl}|jjjd|ddd}t|}|sdsz#split_by_worker..rcsi|]\}}jj||qSr)cat categories)rxcodeZshard)r%rr r{js z#split_by_worker..)numpyrmergercodesrenamerCZTablersort_byasarrayselectZdropwhere concatenatetoolzsliding_windowr;rziprqrrr}r~) rr'r%nprZnrowsrsplitsreZ unique_codesrr)rr%r rDs8   &   rr)rr'rTcsddl}|g|}||||gd}||dd|ddkdd}|dg|g}fddt d|D}| j |dddt ttt |kstt |t |ksttt||S) zL Split data into many arrow batches, partitioned by final partition rNrrcs"g|]\}}j|||dqSrrrrrr rsz&split_by_partition..rr)rrruniquesortrrrrrrr;rrCrqrrr~r<r)rr'rZ partitions partitionrrerrr r|ss &  r|)< __future__rrKrrloggingrrH collectionsrcollections.abcrrconcurrent.futuresrtypingrrr r rZ dask.utilsr Zdistributed.corer Zdistributed.protocolr Zdistributed.shuffle._arrowrrrrrZdistributed.shuffle._commsrZdistributed.shuffle._diskrZdistributed.shuffle._limiterrZdistributed.shuffle._shufflerZdistributed.utilsrrr7rOrrrrr getLoggerrrrrr!rrr|rrrr sF             /