&`ip.dZddlZddlmZmZmZmZmZmZddl Z ddl m Z ddl m Z eZ eZeZGddZGdd eZd ee d ed eeee ffd Ze jGddZdZeeddddeegee fdededeeeee efgefdeee egeefdededed eefdZdZ d%d"Zd#Zed$kr edSdS)&aA simple distributed shuffle implementation in Ray. This utility provides a `simple_shuffle` function that can be used to redistribute M input partitions into N output partitions. It does this with a single wave of shuffle map tasks followed by a single wave of shuffle reduce tasks. Each shuffle map task generates O(N) output objects, and each shuffle reduce task consumes O(M) input objects, for a total of O(N*M) objects. To try an example 10GB shuffle, run: $ python -m ray.experimental.shuffle --num-partitions=50 --partition-size=200e6 --object-store-memory=1e9 This will print out some statistics on the shuffle execution such as: --- Aggregate object store stats across all nodes --- Plasma memory usage 0 MiB, 0 objects, 0.0% full Spilled 9487 MiB, 2487 objects, avg write throughput 1023 MiB/s Restored 9487 MiB, 2487 objects, avg read throughput 1358 MiB/s Objects consumed by Ray tasks: 9537 MiB. Shuffled 9536 MiB in 16.579771757125854 seconds N)AnyCallableIterableListTupleUnion) ObjectRef)Clusterc@eZdZdZdZdeddfdZdeefdZ dS)ObjectStoreWriteraTThis class is used to stream shuffle map outputs to the object store. It can be subclassed to optimize writing (e.g., batching together small records into larger objects). This will be performance critical if your input records are small (the example shuffle uses very large records, so the naive strategy works well). cg|_dSNresultsselfs l/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/experimental/shuffle.py__init__zObjectStoreWriter.__init__2  itemreturnNc^|jtj|dS)zQueue a single item to be written to the object store. This base implementation immediately writes each given item to the object store as a standalone object. N)rappendrayputrrs raddzObjectStoreWriter.add5s( CGDMM*****rc|jS)z6Return list of object refs representing written items.rrs rfinishzObjectStoreWriter.finish=s |r) __name__ __module__ __qualname____doc__rInTyperrr r rrr r )sn++4++++Yrr c<eZdZdZdeddfdZdeefdZdS)ObjectStoreWriterNonStreamingcg|_dSrrrs rrz&ObjectStoreWriterNonStreaming.__init__CrrrrNc:|j|dSr)rrrs rrz!ObjectStoreWriterNonStreaming.addFs D!!!!!rc|jSrrrs rr z$ObjectStoreWriterNonStreaming.finishIs |r) r!r"r#rr%rrrr r&rrr(r(Bsd""4""""S rr( input_streamnum_partitionsrc#8Kd}|D]}||fV|dz }||z}dS)a8Round robin partitions items from the input reader. You can write custom partitioning functions for your use case. Args: input_stream: Iterator over items from the input reader. num_partitions: Number of output partitions. Yields: Tuples of (partition id, input item). rNr&)r,r-irs rround_robin_partitionerr1MsH A$i Q ^rc eZdZdZdZdZdS)_StatusTrackerc>d|_d|_g|_g|_dSNr)num_map num_reducemap_refs reduce_refsrs rrz_StatusTracker.__init__ds%  rc"||_||_dSr)r8r9)rr8r9s rregister_objectrefsz"_StatusTracker.register_objectrefsjs  &rc|jrUtj|jdt|jd\}|_|xjt|z c_n[|jrTtj|jdt|jd\}|_|xjt|z c_|j|jfS)Nr/F)timeout num_returns fetch_local)r8rwaitlenr6r9r7)rreadys r get_progressz_StatusTracker.get_progressns = *#&8  ..! $$$ E4= LLCJJ &LLL   *&)h  011! ''' #E4# OOs5zz )OO|T_,,rN)r!r"r#rr;rCr&rrr3r3bsA '''-----rr3cddlm}d}d}||d}|d||d}|d||ks||krtj|j\}} |||z || |z |}| }tjd||k|||k| | dS)Nr)tqdm)totalpositionz Map Progress.r/zReduce Progress.g?) rEset_descriptionrgetrCremoteupdatetimesleepclose) trackerinput_num_partitionsoutput_num_partitionsrEr6r7map_bar reduce_bar new_num_mapnew_num_reduces rrender_progress_barrVs2GJd-:::G O,,,1A>>>J1222 ( ( (J9N,N,N&)gg.B.I.I.K.K&L&L# ^{W,---.:5666#  3 ( ( (J9N,N,N MMOOOrT) partitionerobject_store_writerrO streaming input_readerrPrQ output_writerrWrXrOrYc | tjdtdtttt t fffd tjdtdtttt t fdtffd fdtD  fdtD}|r6|j d D|t|tj |S) aSimple distributed shuffle in Ray. Args: input_reader: Function that generates the input items for a partition (e.g., data records). input_num_partitions: The number of input partitions. output_num_partitions: The desired number of output partitions. output_writer: Function that consumes a iterator of items for a given output partition. It returns a single value that will be collected across all output partitions. partitioner: Partitioning function to use. Defaults to round-robin partitioning of input items. object_store_writer: Class used to write input items to the object store in an efficient way. Defaults to a naive implementation that writes each input record as one object. tracker: Tracker actor that is used to display the progress bar. streaming: Whether or not if the shuffle will be streaming. Returns: List of outputs from the output writers. r>r0rcfdtD}|D] \}}|||!d|DS)Nc$g|] } Sr&r&).0_rXs r z7simple_shuffle..shuffle_map..s#OOOQ&&((OOOrc6g|]}|Sr&)r )r`cs rrbz7simple_shuffle..shuffle_map..s ,,,q ,,,r)ranger)r0writersout_irrZrXrQrWs r shuffle_mapz#simple_shuffle..shuffle_maps|OOOO%8M2N2NOOO&;||A8MNN % %KE4 EN  t $ $ $ $,,G,,,,rmapper_outputscg}t|ksJ|D]}|D]}||||Sr)rAr)r0ri input_objectsobj_refsobj_refrPr[s rshuffle_reducez&simple_shuffle..shuffle_reducesp >""&:::::& . .H# . .$$W---- .}Q ...rc:g|]}|Sr&)rJ)r`r0rhs rrbz"simple_shuffle..s'RRR{))!,,RRRrc bg|]*jgfdtDR+S)c,g|]}|Sr&r&)r`r0jshuffle_map_outs rrbz-simple_shuffle...s#LLL1#A&LLLr)rJre)r`rrrPrsrns @rrbz"simple_shuffle..sk    LLLLL6J0K0KLLL   rcg|] }|d S)rr&)r`map_outs rrbz"simple_shuffle..s 7 7 7GWQZ 7 7 7r) rrJ PartitionIDrrrr OutTyperer;rVrI) rZrPrQr[rWrXrOrYshuffle_reduce_outrhrsrns `````` @@@rsimple_shufflerysF Z1222-{-tDsI~1F,G'H--------32-  Z/ /)-d5i3H.I)J/ //////Z/SRRRe.input_reader sX~&& U UA''-?C28'TT T T T T U Urshuffle_inputscd}s,|D](}tj|}||j|jzz })nE|rCtj|d\\}}tj|}||j|jzz }|C|S)Nrr/r])rrIsizeitemsizer@)r0rrFrmarrrBuse_waits rr[zrun..output_writers 1) 1 1gg&&CL00 1! 1*-(>q*Q*Q*Q'gennCL00! 1  rc8d}|D]}||j|jzz }|Sr5)rr)r0rrFrs routput_writer_non_streamingz(run..output_writer_non_streamings2! - -C SX , ,EE r)rZrPrQr[rXrOg?r) stats_onlyr/ShuffledizMiB inseconds)rLnumpyprintrinitrrintr3rJrvrr%rr rwrr(r ryrMre_private internal_apimemory_summary Exceptionsum) ray_addressr}r-partition_sizerr| no_streamingrrOrL is_multi_noderrZr[routput_writer_callablerXstart output_sizesdeltasummaryr0rrs ` ` @@rrunrs6KKKM M 3444 $?????  M &''' 'I''((( -8--... D/BDDEEE 85HII ))))) &''' (8KLLLL((N#N'E2 ''))HU U(8UUUUUUUU  d9o '       (,S  /!<;/!. DIKKE!!+,,/ L DIKK% EDJsOOO GGGG 1XX l/>>$>OOGG    DJqMMM D    E  'NNN GGG CL))[9::HeYs*%HH-,H-c ddl}|}|dtd|dtd|dt d|dtd |d t d|d t d |d dd|ddd|}t|j|j |j |j |j |j |j|jdS)Nrz --ray-address)typedefaultz--object-store-memoryrz--num-partitionsrz--partition-sizerz --num-nodesz --num-cpusrz--no-streaming store_trueF)actionrz --use-wait)rr}r-rrr|rr)argparseArgumentParser add_argumentstrfloatr parse_argsrrr}r-rrr|rr)rparserargss rmainrJsWOOO  $ $ & &F c4@@@ /eSIII *a@@@ *FFF  C>>>  3::: (uMMM  \5III     D$ 4**.&      r__main__) NrrrNrFFN)r$rLtypingrrrrrrrr ray.cluster_utilsr r%rwrrvr r(r1rJr3rVboolryrrrr!r&rrrs0 >>>>>>>>>>>>>>>> %%%%%%   2$56"47 eK'()*------- ->: ->"D'D'D'K=(6*::;D'D' D' [$uY^/D*EFOP D'  & 3+!66 D'+D'D'D' ']D'D'D'D'N   aaaaH4 zDFFFFFr