&`i'ddlZddlZddlZddlZddlmZddlmZddlmZm Z ddl m Z m Z m Z mZmZmZmZmZddlmZddlZddlZddlmZddlmZdd lmZdd lmZdd l m!Z!m"Z"m#Z#m$Z$m%Z%dd l&m'Z'm(Z(m)Z)dd l*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1ddl2m3Z3m4Z4m5Z5m6Z6ddl7m8Z8ej9e:Z;dZe e?e e-fZ@deAde?fdZBd/deCdeCde?fdZDGddZEGdd ZFejGd!Gd"d#ZHdeeHfd$ZIGd%d&ZJGd'd(ZKe1eGd)d*ZLeGd+d,ZMeGd-d.ZNdS)0N) defaultdict)contextmanager) dataclassfields)AnyDictListMappingOptionalSetTupleUnion)uuid4) ActorHandle) BlockList) DatasetState)RuntimeMetricsHistogram) NODE_UNKNOWN MetricsGroup MetricsType NodeMetricsOpRuntimeMetrics)DatasetMetadataTopologyget_dataset_metadata_exporter)capfirst) BlockStats) DataContext) DeveloperAPI)CounterGauge HistogramMetric)NodeAffinitySchedulingStrategydatasets_stats_actor_dataset_stats_actorunknownsecondsreturnc|dkr tt|ddzS|dkr#tt|dzddzStt|dzdzddzS)NsgMbP?msus)strround)r(s l/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/stats.pyfmtr48sx{{5!$$%%++ 554++,,t3354$.2233d::lvlspaces_per_indentcd|z|zS)zReturns a string of spaces which contains `level` indents, each indent containing `spaces_per_indent` spaces. For example: >>> leveled_indent(2, 3) ' '  )r7r8s r3leveled_indentr<As # #s **r5cpeZdZdZdZed dZdeddfdZdefdZ defd Z defd Z defd Z dS) Timerz8Helper class for tracking accumulated time (in seconds).cXd|_td|_d|_d|_dS)Nrinf)_totalfloat_min_max _total_countselfs r3__init__zTimer.__init__Ms+  <<  #$r5r)Nc#Ktj} dV|tj|z dS#|tj|z wxYwN)time perf_counteradd)rG time_starts r3timerz Timer.timerSsn&((  7 EEE HHT&((:5 6 6 6 6 6DHHT&((:5 6 6 6 6s A+A1valuec|xj|z c_||jkr||_||jkr||_|xjdz c_dS)Nr+)rArCrDrE)rGrPs r3rMz Timer.add[sT u 49  DI 49  DI Qr5c|jSrJ)rArFs r3getz Timer.getcs {r5c|jSrJ)rCrFs r3minz Timer.minf yr5c|jSrJ)rDrFs r3maxz Timer.maxirVr5cL|jr|j|jz ntdS)Nr@)rErArBrFs r3avgz Timer.avgls%262CUt{T...uUr5r)N) __name__ __module__ __qualname____doc__rHrrOrBrMrSrUrXrZr;r5r3r>r>JsBB%%% 777^74UUUVUVVVVVVr5r>cReZdZdZdedddeefdZdeddfd Z d e ddfd Z d S) _DatasetStatsBuilderzHelper class for building dataset stats. When this class is created, we record the start time. When build() is called with the final blocks of the new dataset, the time delta is saved as part of the stats. operator_nameparent DatasetStatsoverride_start_timecV||_||_|ptj|_dSrJ)rbrcrKrL start_time)rGrbrcres r3rHz_DatasetStatsBuilder.__init__ws, + -D1B1D1Dr5metadatar)ci}t|D]o\}\}}t|}t|dkr;|dkr|||j|z<>|||jdd|z<e|||j<pt ||j|j}tj |j z |_ |S)Nr+rz->)rhrc base_name) enumerateitemsrlenrbsplitrdrcrKrLrg time_total_s)rGrh op_metadataikvcapped_kstatss r3build_multioperatorz(_DatasetStatsBuilder.build_multioperators "8>>#3#344 4 4IAv1{{H8}}q  66ABK 2X =>>QRK 2 8 8 > >r BX MNN23 D.// ;(    ".004?B r5 final_blocksct|j|i|j}t j|jz |_|S)N)rhrc)rdrb get_metadatarcrKrLrgrp)rGrxrvs r3buildz_DatasetStatsBuilder.buildsQ(,*C*C*E*EF;   ".004?B r5N) r\r]r^r_r1r rBrH StatsDictrwrr{r;r5r3raraps## EEE&e_ EEEEI.&)r5ra)num_cpusceZdZdZd$dZdedeedfdeee ffdZ deee ffd Z defd Z d%d ed eeeeeeffdeedeeefdeeeeeeeeffff dZd&dZ d'dZded edeededef dZd edeeeffdZd%deefdZdedefdZdedeeeffd Z d(d ed!eed"eefd#Zd S)) _StatsActoraActor holding stats for blocks created by LazyBlockList. This actor is shared across all datasets created in the same cluster. In order to cap memory usage, we set a max number of stats to keep in the actor. When this limit is exceeded, the stats will be garbage collected in FIFO order. TODO(ekl) we should consider refactoring LazyBlockList so stats can be extracted without using an out-of-band actor.r.c tjt|_i|_i|_||_d|_i|_i|_ t|_ i|_ tj |_d}tdd||_tdd||_tdd ||_td d ||_td d ||_tdd||_tdd||_|t0j||_|t0j||_|t0j||_|t0j||_ |t0j!||_"|t0j#||_$|%|_&d}tdd||_'tdd||_(tdd||_)tdd||_*tdd||_+tdd||_,tdd ||_-td!d"||_.td#d$||_/td%d&||_0td'd(||_1td)d*||_2td+d,||_3td-d.||_4td/d0||_5td1d2||_6td3d4||_7td5d6||_8td7d8||_9d9}td:d;||_:tdd?d@<dAtzDdB||_>d}tdCdD||_?tdEdF||_@tdGdH||_AtdIdJd@<dKtzDdB||_BdS)LNrdatasetoperatordata_spilled_byteszBytes spilled by dataset operators. DataContext.enable_get_object_locations_for_metrics must be set to True to report this metric descriptiontag_keysdata_freed_bytesz Bytes freed by dataset operatorsdata_current_bytesz9Bytes currently in memory store used by dataset operatorsdata_cpu_usage_coresz#CPUs allocated to dataset operatorsdata_gpu_usage_coresz#GPUs allocated to dataset operatorsdata_output_bytesz$Bytes outputted by dataset operatorsdata_output_rowsz#Rows outputted by dataset operators) metrics_groupr)r%data_iter_time_to_first_batch_secondszTotal time spent waiting for the first batch after starting iteration. This includes the dataset pipeline warmup time. This metric is accumulated across different epochs. data_iter_block_fetching_secondsz>Seconds taken to fetch (with ray.get) blocks by iter_batches()data_iter_batch_shaping_secondszCSeconds taken to shape batch from incoming blocks by iter_batches()"data_iter_batch_formatting_secondsz1Seconds taken to format batches by iter_batches()!data_iter_batch_collating_secondsz2Seconds taken to collate batches by iter_batches()"data_iter_batch_finalizing_secondsdata_iter_total_blocked_secondsz0Seconds user thread is blocked by iter_batches()data_iter_user_secondszSeconds spent in user codedata_iter_initialize_secondsz-Seconds spent in iterator initialization code!data_iter_get_ref_bundles_secondsz:Seconds spent getting RefBundles from the dataset iteratordata_iter_get_secondsz;Seconds spent in ray.get() while resolving block referencesdata_iter_next_batch_secondsz:Seconds spent getting the next batch from the block bufferdata_iter_format_batch_secondsz"Seconds spent formatting the batchdata_iter_collate_batch_secondsz!Seconds spent collating the batch data_iter_finalize_batch_secondsz"Seconds spent finalizing the batchdata_iter_blocks_localz*Number of blocks already on the local nodedata_iter_blocks_remotez8Number of blocks that require fetching from another nodedata_iter_unknown_locationz,Number of blocks that have unknown locationsdata_iter_prefetched_bytesz2Current bytes of prefetched blocks in the iteratorrjob_idrg#data_dataset_estimated_total_blocksz&Total work units in blocks for dataset!data_dataset_estimated_total_rowsz$Total work units in rows for datasetdata_dataset_statezState of dataset (z, c0g|]}|jd|jS=rPname.0r-s r3 z(_StatsActor.__init__..s+7d7d7dRS178M8MQV8M8M7d7d7dr5)$data_operator_estimated_total_blocksz'Total work units in blocks for operator"data_operator_estimated_total_rowsz%Total work units in rows for operatordata_operator_queued_blocksz$Number of queued blocks for operatordata_operator_statezState of operator (c0g|]}|jd|jSrrrs r3rz(_StatsActor.__init__..s+8e8e8eSTAG9N9Naf9N9N8e8e8er5)C collectionsrdictrh last_timerg max_statsnext_dataset_iddatasets_ray_nodes_cacher_metadata_exporterdataset_metadatasdequefinished_datasets_queuer! spilled_bytes freed_bytes current_bytescpu_usage_coresgpu_usage_cores output_bytes output_rows0_create_prometheus_metrics_for_execution_metricsrINPUTSexecution_metrics_inputsOUTPUTSexecution_metrics_outputsTASKSexecution_metrics_tasksOBJECT_STORE_MEMORY"execution_metrics_obj_store_memoryACTORSexecution_metrics_actorsMISCexecution_metrics_misc/_create_prometheus_metrics_for_per_node_metricsper_node_metricstime_to_first_batch_siter_block_fetching_siter_batch_shaping_siter_batch_formatting_siter_batch_collating_siter_batch_finalizing_siter_total_blocked_s iter_user_siter_initialize_siter_get_ref_bundles_s iter_get_siter_next_batch_siter_format_batch_siter_collate_batch_siter_finalize_batch_siter_blocks_localiter_blocks_remoteiter_unknown_locationiter_prefetched_bytesrrjoinrrrrrr)rGr op_tags_keys iter_tag_keys dataset_tags operator_tagss r3rHz_StatsActor.__init__s^#/55 " !(* 13#@"A"A=?(3'8':':$ / # ="    ! :!    # S!    % "=!    % "=!    " >!    ! =!     A A*1% B   %  A A*2% B   &  A A*0% B   $  A A*>% B   /  A A*1% B   %  A A*/% B   #!% T T V V$ %* 3r" & & & "&+ .X"& & & " %* -]"% % % ! (- 0K"( ( ( $ ', /L"' ' ' # (- 0L"( ( ( $ %* -J"% % % ! ! $4"    "' *G"" " "  ', /T"' ' ' #  #U"    "' *T"" " "  $) ,<"$ $ $  %* -;"% % % ! &+ .<"& & & " "' $D"" " "  #( %R"# # #  &+ (F"& & & " &+ (L"& & & "; 38 1@!4 4 4 0 27 />!2 2 2 . #( hTYY7d7dWc7d7d7d-e-ehhh!# # #  0 49 2A"5 5 5 1 38 0?"3 3 3 / ,1 )>", , , ( $) !idii8e8eXd8e8e8e.f.fiii"$ $ $    r5rr.r)ci}tjD]}|j|ksd|j}|j}|jt jkrt|||||j<O|jt jkrt|f||d|j ||j<|jt j krt|||||j<|S)Ndata_r) r get_metricsrrr metrics_typerr!r" metrics_argsr )rGrrmetricsmetric metric_namemetric_descriptions r3rz<_StatsActor._create_prometheus_metrics_for_execution_metricss &244  F'=88/&+//K!'!3 "k&777', 2%((( $$ $ (==='0( 2%(() (( $$ $ (;;;'. 2%((( $ r5ci}ttD]'}d|jd}t|dd||j<(|S)Nr _per_node)rnode_ipr)rrrr!)rGrfieldrs r3rz;_StatsActor._create_prometheus_metrics_for_per_node_metricss^K((  E7%*777K"'/###GEJ   r5cNt|j}|xjdz c_|S)z3Generate a unique dataset_id for tracking datasets.r+)r1r)rG dataset_ids r3gen_dataset_idz_StatsActor.gen_dataset_ids,-..  !r5N dataset_tag op_metricsrstaterc  ddtdtttttfdt t t ffd}t||D]\}}|||} |j | dd| |j | dd| |j | dd| |j | d d| |j | d d| |j | d d| |j | d d| |jD]&\} } || | | d| '|jD]&\} } || | | d| '|jD]&\} } || | | d| '|jD]&\} } || | | d| '|jD]&\} } || | | d| '|jD]&\} } || | | d| '||D]\} } | |jvr||j | t6}||| } | D]\}}|j|} || ||  |||dS)N prom_metricrPtagscBt|tr|||dSt|tr|||dSt|t r-t|t r|||dSdSdSrJ) isinstancer!setr incr"r export_to)rrPrs r3_recordz5_StatsActor.update_execution_metrics.._records +u-- 7t,,,,,K11 7t,,,,,K33 7e%<==7OOK66666 7 777r5obj_store_mem_spilledrobj_store_mem_freedobj_store_mem_usedbytes_task_outputs_generatedrow_outputs_taken cpu_usage gpu_usage)r node_ip_tagrJ)r#rintrBr rr1zip _create_tagsrrrSrrrrrrrrmrrrrrr_rebuild_ray_nodes_cacherrupdate_dataset)rGrrrrrr rv operator_tagr field_namernode_id node_metricsrr metric_values r3update_execution_metricsz$_StatsActor.update_execution_metricssI$( 7 7 7eT#Y./ 7sCx. 7 7 7 7$'z=#A#A E E E<$$[,??D   " "599-Da#H#H$ O O O   +@!!D!Dd K K K   " "599-A1#E#Et L L L   ! !%)),JA"N"NPT U U U   +>!B!BD I I I  $ $UYY{A%>%> E E E  $ $UYY{A%>%> E E E+/+H+N+N+P+P E E' K UYYz1%=%=tDDDD+/+I+O+O+Q+Q E E' K UYYz1%=%=tDDDD+/+G+M+M+O+O E E' K UYYz1%=%=tDDDD 8>>@@ E E UYYz1%=%=tDDDD+/+H+N+N+P+P E E' K UYYz1%=%=tDDDD+/+F+L+L+N+N E E' K UYYz1%=%=tDDDD E  ')9)?)?)A)A = =%$"77711333/33G\JJ(([g(VV1=1C1C1E1E==-K"&"7 "DKGKt<<<<= K/////r5ctj}|D]<}|dd}|dd}| | ||j|<=dS)NNodeIDNodeName)raynodesrSr)rG current_nodesnoder node_names r3rz$_StatsActor._rebuild_ray_nodes_cache'se ! ; ;Dhhx..GT22I"y'<1:%g.  ; ;r5rvrdc ||}|j|j||j|j||j|j||j|j||j|j||j|j||j |j ||j |j ||j |j ||j |j ||j |j ||j|j||j|j||j|j||j|j||j|j ||j|j||j|j||j|j|dSrJ)rrrrSrrrrrrrrrrrrrrrriter_time_to_first_batch_srr)rGrvrrs r3update_iteration_metricsz$_StatsActor.update_iteration_metrics/s   -- ""5#:#>#>#@#@$GGG #''(D(H(H(J(JDQQQ E,0022D999 ""5#:#>#>#@#@$GGG  $$U%>%B%B%D%DdKKK !%%e&@&D&D&F&FMMM "&&u'B'F'F'H'H$OOO ""5#:DAAA ##E$z0_StatsActor.register_dataset..^sF )16 !%& r5)rrr-r. total_rowsrgend_time operators)rr)rrgr*execution_start_timeexecution_end_timer) rKrr0rrrrrexport_dataset_metadata)rGrrrr)r*rgs r3register_datasetz_StatsActor.register_datasetMsY[[ !).$!.& &  k"$  " .2A!&%)%)#'"*/ 3 3 3D "; /  # ; ;&{3      / .r5c|j|||j|}|j|dd}tt |j|dd}|||d}|j|dd||j|dd||dtj j }tj |}|j |j ||||i}|d iD]\} } || d } |j| dd| |j| dd| |j| d d| | dtj j }tj |}|j|j | ||| <||||dtjj tjj hvr|j|t3|j|jkr~|jr{|j} |j| d|j| dt3|j|jkr |jsdSdSdSdSdS) NrNonergrrr.r2rr4rr/)rupdaterSr1rrrrrUNKNOWNr from_stringrrPupdate_dataset_metadata_statermrrrr'update_dataset_metadata_operator_statesFINISHEDFAILEDrappendrnrpopleftpopr) rGrrrrgr state_string state_enumoperator_statesrop_stater tag_to_evicts r3rz_StatsActor.update_datasetwse k"))%000 k*{+//&AAT];7;;L!LLMMNN #$  044 IIgq ! !<    .22 IIlA & &    yy,*>*CDD !-l;;  ##J$4lCCC **; EEE+-"'))K"<"<"B"B"D"D 5 5 Hh&$M  5 9 9 Wa((-     3 7 7 \1--}     , 0 0 _a00-    $<<1E1JKKL%1,??J  $ ( ()9= I I I(4OH % % 44[/RRR >l38,:M:RS S S  ( / / < < <dm$$t~55$:V5#;CCEE  !!,555&**<>>>dm$$t~55$:V555 T S555555r5c`s|jSfd|jDS)Nc4i|]\}}|dk||S)rr;)rrsrtrs r3r1z,_StatsActor.get_datasets..s-PPPA!H+:O:O1:O:O:Or5)rrm)rGrs `r3 get_datasetsz_StatsActor.get_datasetss; != PPPP!4!4!6!6PPPPr5r new_statec ||jvrdStj}|j|}|j|krdStj|}||_|t jjkr||_nb|t j jt j jfvr>||_ |j j D]*}|jt jjkr||_||_ +||j|<|j|j|dddSdS)NFinclude_data_contextinclude_op_args)rrKrcopydeepcopyrRUNNINGrr5r@rAr6r)r4rr7)rGrrM update_timedataset_metadataupdated_dataset_metadatars r3r>z)_StatsActor.update_dataset_metadata_states& T3 3 3 Fikk 1*=  !Y . . F#'=1A#B#B )2 &  ,1 1 1 >>\%9%>>>%.HN2=H/-Ez*  " .  # ; ;(%* % <      / .r5rGcp||jvrdS|j|}d}|jjD]%}|j|vr|j||jkrd}n&|sdSt j|}tj}|jjD]}|j|vrv||j}|j|kr$||_|tj j kr||_ H|tj j tj j fvr||_|j s||_ ||j|<|j|j|dddSdS)NFTrO)rr)r4idrrRrSrKrrTrr5r@rAr6rr7) rGrrGrV update_neededrrWrUrMs r3r?z3_StatsActor.update_dataset_metadata_operator_statess T3 3 3 F1*= (1;  H ..Nohk&BBB $   F#'=1A#B#B ikk 09C D DH{o--+HK8 >Y..!* 4 9994?H11 ). ',#3>H/$8D8C5-Ez*  " .  # ; ;(%* % <      / .r5rrc*d|i}|||d<|||d<|S)Nrrrr;)rGrrrrs r3rz_StatsActor._create_tagss2 ;'  #+D   ")DO r5)r.rJr[)rvrd)NN)r\r]r^r_rHrr r1rr#rr!rrr rrrBrr rrr(rrr8rrLr>r?rr;r5r3rrs55C C C C J)5:38_ c6k < c5jAQ    OS H0H0H0c5e#4456H0Cy H0 CH~ H0 #4T#uS%Z7H2H-I(I#JK H0H0H0H0T;;;;<<<<<<(((Cy (  ( " ((((T7?#7?d38n7?7?7?7?rQQ8C=QQQQ 6--04S#X----d'+%)   sm c]      r5rctjjjt dtjjjj}t d|ttj d}t ttdd|S) zEach cluster will contain exactly 1 _StatsActor. This function returns the current _StatsActor handle, or create a new one if one does not exist in the connected cluster. The _StatsActor is pinned on on driver process' node. NzEGlobal node is not initialized. Driver might be not connected to Ray.z"Stats Actor located on cluster_id=F)softTdetached)r namespace get_if_existslifetimescheduling_strategy)r!_privateworker _global_node RuntimeError cluster_idloggerdebugr$get_runtime_context get_node_idroptionsSTATS_ACTOR_NAMESTATS_ACTOR_NAMESPACEremote)current_cluster_idrbs r3get_or_create_stats_actorrq s  |'/ S   ,9D LLJ6HJJKKK9 !!--//     '/     fhh r5c :eZdZdZedeedeee ee e e e ffffdZ ede deedee dee effdZed d de fd Zede dee d ed efdZede fdZdS) _StatsManageraA Class containing util functions that manage remote calls to _StatsActor. Ray Data updates metrics through the _StatsManager, and direct remote calls to the _StatsActor is discouraged. Some functionalities provided by _StatsManager: - Format and update iteration metrics - Format and update execution metrics - Aggregate per node metrics - Dataset registration rr)c 0tjjsdStd}|D]h}|jD]L\}}||}t tD]*}||jxxt||jz cc<+Mi|S)aR Aggregate per-node metrics from a list of OpRuntimeMetrics objects. If per-node metrics are disabled in the current DataContext, returns None. Otherwise, it sums up all NodeMetrics fields across the provided metrics and returns a nested dictionary mapping each node ID to a dict of field values. Nc*ttSrJ)rrr;r5r3z;_StatsManager._aggregate_per_node_metrics..EsS1A1Ar5) r get_currentenable_per_node_metricsr_per_node_metricsrmrrrgetattr)raggregated_by_noderrragg_node_metricsfs r3_aggregate_per_node_metricsz)_StatsManager._aggregate_per_node_metrics7s&((@ 4()A)ABB! N NG)0)B)H)H)J)J N N%#5g#>  ,,NNA$QV,,, af0M0MM,,,,N N "!r5rrrct|}d|D}|||||f} tjj|dS#t $r*}t d|dYd}~dSd}~wwxYw)Nc6g|]}|Sr;)as_dict)rrs r3rz:_StatsManager.update_execution_metrics..Vs"FFFFNN,,FFFr5zKError occurred during update_execution_metrics.remote call to _StatsActor: Texc_info)rsr~rqrro Exceptionrhwarning)rrrrrop_metrics_dictsargses r3rz&_StatsManager.update_execution_metricsNs)DDZPPFF:FFF         G % ' ' @ G N N N N    NNa^_aa     FFFFF  sA BA;;Brvrdc||f} tjj|dS#t$r*}td|dYd}~dSd}~wwxYw)NzKError occurred during update_iteration_metrics.remote call to _StatsActor: Tr)rqr(rorrhr)rvrrrs r3r(z&_StatsManager.update_iteration_metricsgs{#  G % ' ' @ G N N N N    NNa^_aa           s# AAAr)r*ctjtj||||dS)a1Register a dataset with the stats actor. Args: dataset_tag: Tag for the dataset operator_tags: List of operator tags topology: Optional Topology representing the DAG structure to export data_context: The DataContext attached to the dataset N)rqr8ror!rj get_job_id)rrr)r*s r3register_dataset_to_stats_actorz-_StatsManager.register_dataset_to_stats_actorrsS "##4;;  # % % 0 0 2 2          r5c t}tj|jS#t $r:}t d|tj cYd}~Sd}~wwxYw)Nz?Failed to generate dataset_id, falling back to random uuid_v4: ) rqr!rSrrorrhrrhex) stats_actorrs r3gen_dataset_id_from_stats_actorz-_StatsManager.gen_dataset_id_from_stats_actors 355K7;5<<>>?? ?    NNURSUU    77;        s8; A?/A:4A?:A?N)r\r]r^r_ staticmethodr rr r r1rrrBr~rrrr(rrrrr;r5r3rsrs+si  ")*" '#wsE#u*,='=>>? @"""\",)*CyCH~ \0S\  Cy  "    \ . S   \   r5rsceZdZdZdddedeededfdefdZ e dZ dd ed ee d e fd ZddZd efdZdS)rdzHolds the execution times for a given Dataset. This object contains a reference to the parent Dataset's stats as well, but not the Dataset object itself, to allow its blocks to be dropped from memory.N)rkrhrcrkcf||_|t|ts|g}|pg|_|jsdn t d|jDdz|_||_d|_d|_t|_ t|_ t|_ t|_ t|_t|_t|_t|_t|_t|_t|_t|_t|_i|_d|_d|_d|_d|_d|_d|_d|_t|_dS)aCreate dataset stats. Args: metadata: Dict of operators used to create this Dataset from the previous one. Typically one entry, e.g., {"map": [...]}. parent: Reference to parent Dataset's stats, or a list of parents if there are multiple. base_name: The name of the base operation for a multi-operator operation. Nrc3$K|] }|jV dSrJ)numberrps r3 z(DatasetStats.__init__..s$*J*J18*J*J*J*J*J*Jr5r+ unknown_uuid) rhrlistparentsrXrrk dataset_uuidrpr>streaming_exec_schedule_s iter_wait_srrrrrrr'rrr iter_total_s extra_metricsrrrrglobal_bytes_spilledglobal_bytes_restoreddataset_bytes_spilledstreaming_split_coordinator_s)rGrhrcrks r3rHzDatasetStats.__init__sw"$,  j&>&> XF-3\r \ NAAs*J*JT\*J*J*J'J'JQ'N #"0#$16&#(''-2WW#!&(-*/'' +077!,1GG"16'+077!"'''(-#(77'('(*+"*+"*+!*+"*+"5:GG***r5ctSrJ)rqrFs r3rzDatasetStats.stats_actors(***r5rrer)c$t|||S)z>Start recording stats for an op of the given name (e.g., map).)ra)rGrres r3 child_builderzDatasetStats.child_builders$D$0CDDDr5DatasetStatsSummarycn g}t|jdk t|j|j|j|j|j|j|j |j |j |j |j |j|j|j|j|j|j}g}|jd|jD}d}t+|D]\}}|jru|jd}t/|jt2r|jddnd}t6d|dzd|jd |d ||z } fd |jD} t+| D]a\}} r>|dkr|| _n7| |dz } | jrd| jvr | jdnd| _n|| _| | b|j!r|j!nd} tE||||j#|j$|j%|j&|j'|j(|j)|j*| S) zGenerate a `DatasetStatsSummary` object from the given `DatasetStats` object, which can be used to generate a summary string.r+Nc6g|]}|Sr;) to_summaryrs r3rz+DatasetStats.to_summary.. s $J$J$JQ\\^^$J$J$Jr5rrjsumzParent z (operator: z) contributes z rows to inputcPg|]"\}}t||#S))is_sub_operator)OperatorStatsSummaryfrom_block_metadata)rrrvrs r3rz+DatasetStats.to_summary..sK   e ! 4 4e_ 5     r5)+rnrhIterStatsSummaryrrrrrrrr'rrrrrrrrrrrloperators_statsroutput_num_rowsrrSrhrirbrmtotal_input_num_rowsrBrrrrrprkrrrr)rGr iter_statsstats_summary_parentsparent_total_outputrrparent_summarylast_parent_op op_outputop_statsop_statprev_oprrs @r3rzDatasetStats.to_summarysdm,,q0%    ' O  "  $  %  &  +  %    "    .  "  #  &  &#  (!# < #$J$JT\$J$J$J ! !*+@!A!A 1 1 A~- 1!/!?!C"."@$GGN266ua@@@  va!evv1Mvv]fvvv$y0#     $}2244    $H-- , ,JAw C663FG00'q1uoG$38=AX8X8X /66 000C,  " "7 + + + +- D * . . 0 0 0 " #   ! K     N    %  &  & %   r5cN|S)axGenerate a string representing the runtime metrics of a Dataset. This is a high level summary of the time spent in Ray Data code broken down by operator. It also includes the time spent in the scheduler. Times are shown as the total time for each operator and percentages of time are shown as a fraction of the total time for the whole dataset.)rruntime_metricsrFs r3rzDatasetStats.runtime_metricsLs   00222r5rJ)r)r)r\r]r^r_r|rr r r1rHpropertyrrBrarrrr;r5r3rdrds A<A<A<A<h~.^0DDE A<  A<A<A<A| | |t|z }ftjj} | r6|jr/|r |jrdnd} || z }|dt|jzdzz }|t|jz }t|jdkr|rt|jdz }t|jdz }|s|r5|dz }|d|z }|d|z }t|jdz }|r|dz }|d|z }|jr|dz }|dz }|d|jdz }| r|r|d|zz }|S)aReturn a human-readable summary of this Dataset's stats. Args: already_printed: Set of operator IDs that have already had its stats printed out. include_parent: If true, also include parent stats summary; otherwise, only log stats of the latest operator. add_global_stats: If true, includes global stats to this summary. Returns: String with summary statistics for executing the Dataset. NrF)add_global_stats r+rzOperator {} {}: z[execution cached] r,z Operator {} {}: executed in {}s z Suboperator {} {}: z [execution cached]  z* Extra metrics: .Az Cluster memory: z* Spilled to disk: {}MB z* Restored from disk: {}MB z Dataset memory: zDataset throughput:  * Ray Data throughput:  rows/s )rr to_stringrnrrbrformatrrMr1r2rprkrlrrwverbose_stats_logsrrrrrrnum_rows_per_sr)rGrrroutr parent_sumoperators_stats_summaryrb operator_uuid rounded_totalnrindent mb_spilled mb_restoreddataset_mb_spilleds r3rzDatasetStatsSummary.to_stringes"  "!eeO < N \  [[5[QQ  :%C4KC"& t# $ $ ) )&*&:1&= #3AM - =M %,,T[-HH HC//--##M222s2333 % & & * *!$"3Q77M!! ! 6== T^] C/88L.M.M 8 8** 7 E $ 1M A t .55aGGG O3333CC#'' 66636777CC(466I  H$"4 H+/F/V  6MC &T-?)@)@@4G GC s4?### t# $ $q ( (-= (t83>??J :S @AAK J[ J,,299*EEE5<<[III!&t'AC'G!H!H ! N,,299:LMMM" Rt ..Q43FQQQQ  1"2 1 4$..000 0C r5c|jr|jdjnd}|r|dnd}|}|r|sdS||z S)zDCalculates the throughput in rows per second for the entire dataset.rjrr)rrget_total_wall_time)rGrtotal_num_out_rows wall_times r3rz"DatasetStatsSummary.num_rows_per_ssm9=8L SD  $ 4 4RS 8GM_U33A,,.. !  3!I--r5currcg}|jD]8}|r4|jr-|t|9||gzSrJ)rextendr _collect_dataset_stats_summaries)rsummsrs r3rz4DatasetStatsSummary._collect_dataset_stats_summariess^ V VA VQY V 0QQRSTTUUUv~r5summctd|jD}td|jD}||fS)Nc3$K|] }|jV dSrJ)earliest_start_timeropss r3rz:DatasetStatsSummary._find_start_and_end..s%UUS4UUUUUUr5c3$K|] }|jV dSrJ)latest_end_timers r3rz:DatasetStatsSummary._find_start_and_end..s%MM,MMMMMMr5)rUrrX)rearliest_start latest_ends r3_find_start_and_endz'DatasetStatsSummary._find_start_and_endsKUU@TUUUUUMM8LMMMMM z))r5c|dtdtdtffd }t|}d}|D]P}t |jdkr6t|\}}||z }|||j|z }Q||d|j z }||dz }|S) NrrKr)cXdkr|z nd}d|dt|d|dzddS)Nrz* : z (dz.3fz%) r4)rrKfractiontotal_wall_times r3fmt_linez5DatasetStatsSummary.runtime_metrics..fmt_linesJ1@11D1Dto--!HEEED EEX^EEEE Er5zRuntime Metrics: r SchedulingTotal) rr1rBrrrnrrrkr) rGr summariesrrrr op_total_timers @r3rz#DatasetStatsSummary.runtime_metricss2244 F3 Fe F F F F F F F(HHNN " ? ?D4'((1,,-@-T-T..* !+^ ; xx >>> xx d&DEEE xx111 r5rct}dfd|jD}dfd|jD}dfd|jD}|r d|d|dnd}|r d|d|dnd}|r d|d|dnd}dg|d|d |jd|d |jd|d |jd|d |d |d|d|d|j dzd|d|j dz d|d|j dz d|d|j dz d|d|d|dS)Nrc@g|]}|dzSr,__repr__)rsslevels r3rz0DatasetStatsSummary.__repr__..s) C C CR[[ # # C C Cr5c@g|]}|dzSrr)rpsrs r3rz0DatasetStatsSummary.__repr__..s)!P!P!PR"++eai"8"8!P!P!Pr5c3RK|]!\}}tdz|d|dV"dS)r,r,N)r<)rrsrtrs r3rz/DatasetStatsSummary.__repr__..s]" " 1eai(( 3! 3 3q 3 3 3" " " " " " r5, z rzDatasetStatsSummary( z dataset_uuid=z base_name=z number=z extra_metrics={z}, z operators_stats=[z], z iter_stats=r+z global_bytes_spilled=rzMB, z global_bytes_restored=z dataset_bytes_spilled=z parents=[r)r<rrrrrmrrkrrrrrr)rGrrr parent_statsrs ` r3rzDatasetStatsSummary.__repr__s&&)) C C C Cd.B C C C  yy!P!P!P!P4.sF   4'((1,, 3 3D 9 9,,,r5rc3&K|] }|dV dS)rNr;r start_ends r3rz:DatasetStatsSummary.get_total_wall_time.."s& J J)1 J J J J J Jr5c3&K|] }|dV dS)r+Nr;r s r3rz:DatasetStatsSummary.get_total_wall_time..#s&FFiYq\FFFFFFr5)rrrnrUrX)rG start_endsrrs r3rz'DatasetStatsSummary.get_total_wall_times   +LLTRR   z??a  1 J Jz J J JJJNFF:FFFFFJ. .r5cht|}td|DS)zGCalculate the sum of the wall times across all blocks of all operators.c3RK|]"}td|jDV#dS)c3^K|](}|jr|jddndV)dSrrN)rrSrs r3rzJDatasetStatsSummary.get_total_time_all_blocks...+sT47=GCM%%eQ///ar5N)rrr s r3rz@DatasetStatsSummary.get_total_time_all_blocks..)sa   #3      r5)rrr)rGrs r3get_total_time_all_blocksz-DatasetStatsSummary.get_total_time_all_blocks&sE'HHNN   "      r5ctd|jD}|td|jDzS)Nc3>K|]}|VdSrJ)get_total_cpu_timers r3rz9DatasetStatsSummary.get_total_cpu_time..4s.FFA--//FFFFFFr5c3LK|]}|jddV dSr)cpu_timerSrrs r3rz9DatasetStatsSummary.get_total_cpu_time..5sC  *,BKOOE1 % %      r5)rrr)rGrs r3rz&DatasetStatsSummary.get_total_cpu_time3sXFFFFFFF C  040D       r5cd|jD}|rt|nd}|js|St|gd|jDRS)Nc6g|]}|Sr;)get_max_heap_memoryrs r3rz;DatasetStatsSummary.get_max_heap_memory..:s$GGGQ..00GGGr5rcDg|]}|jddS)rXr)memoryrSrs r3rz;DatasetStatsSummary.get_max_heap_memory..As( E E E"bimmE1%% E E Er5)rrXr)rG parent_memory parent_maxs r3rz'DatasetStatsSummary.get_max_heap_memory9srGG$,GGG +8?S'''a #    E E0D E E E    r5)NTTr )r\r]r^r __annotations__rr1rBrrr r boolrrrrrr rrrrrrrr;r5r3rrUsE01111"""" ' (((( KKKNNNS>!!!$$$$/3# WW!#c(+WW WWWWr....X."# # $\*"7*E%,>>r5cr|jr|jr |jdsdS|jd|jdz S)zCCalculates the estimated single-task throughput in rows per second.rr)rrrFs r3num_rows_per_task_sz(OperatorStatsSummary.num_rows_per_task_smsF# 4> PUAV 3#E*T^E-BBBr5 block_statscP d|D}d}d}d\}}|r7td|D}td|D}||z }|r#dt|} nE|r.s!TTTq1<;Sal;S;S;Sr5r)rrc3$K|] }|jV dSrJ) start_time_srs r3rz;OperatorStatsSummary.from_block_metadata..s$%I%Ian%I%I%I%I%I%Ir5c3$K|] }|jV dSrJ) end_time_srs r3rz;OperatorStatsSummary.from_block_metadata..s$!C!C1!,!C!C!C!C!C!Cr5z{} blocks produced r,z{} blocks produced in {}srrN)rUrXmeancountz{} tasks executed, {})NNNNcg|] }|j Sr; wall_time_srrs r3rz.>>>aAM>>>r5cg|] }|j Sr;r:r<s r3rz.r=r5cg|] }|j Sr;r:r<s r3rz.s C C C1 C C Cr5cg|] }|j Sr;r:r<s r3rz.r=r5)rUrXr7rcg|] }|j Sr; cpu_time_sr<s r3rz.===QAL===r5cg|] }|j Sr;rBr<s r3rz.rDr5cg|] }|j Sr;rBr<s r3rz. B B B! B B Br5cg|] }|j Sr;rBr<s r3rz.rDr5cBg|]}t|jpddz dS)rir,)r2 max_uss_bytesr<s r3rz.s;EFq+! .rDr5cg|] }|j Sr;rLr<s r3rz.rDr5cg|] }|j Sr;rLr<s r3rz.rGr5cg|] }|j Sr;rLr<s r3rz.rDr5c*g|]}|j |jSrJ)num_rowsr1s r3rz.s!UUU!aj>T1:>T>T>Tr5c*g|]}|j |jSrJ) size_bytesr1s r3rz.s(   1IAL1I1I1Ir5c4i|]\}}|t|Sr;)rn)rr$taskss r3r1z.s$RRR e4URRRr5)rbrrprrr%rrr&rrrr'r(r))rUrXrrnr2rrrrRr0task_idxvaluesnpr7rrrrrMrmr)clsrbr-rr0rrprrexec_summary_strr)metatask_rows_statswall_time_stats cpu_stats memory_stats udf_statsmemory_stats_mboutput_num_rows_statsroutput_size_bytes_statsr'node_counts_stats node_tasksr- node_countsrs r3rz(OperatorStatsSummary.from_block_metadataxs"UTKTTT   /3,_  A#&%I%Ij%I%I%I"I"I !!C!C !C!C!CCCO*-@@L  %5<#E#E OO]$$  $&  $ +C00  E ED}(T_-H$/2333t}D333 y>>A  9++--..9++--..BGD)9)9););$<$<==>>Y O 7==I 0   ?U;L)  >>:>>>??>>:>>>?? C C C C CDD>>:>>>?? O==*===>>==*===>> B Bz B B BCC==*===>> IJTO?++?++BGO4455L==*===>>==*===>> B Bz B B BCC==*===>> I!%UU{UUU  ?++?++BGO4455?++ %% !#'  "-     ,--,--BG$56677,-- '' #!  $055J 6 619%))!*5555RRz?O?O?Q?QRRRK;--//00;--//00BGD););)=)=$>$>??@@[)) !!  $#'+% 3+(8%!515(%    r5c 8|jrdnd}|j}|j}|rl||z }|dt |dt |dt |dt |dz }|j}|rl||z }|dt |dt |dt |dt |dz }|j}|rl||z }|d t |dt |dt |dt |dz }|j}|r1||z }|d |d|d|dz }|j}|r8||z }|d |d|d|d|dz }|j }|r8||z }|d |d|d|d|dz }|j } | r8||z }|d | d| d| d| dz }|j } | r8||z }|d| d| d| d| dz }|j rc|j r\|jr|jnd} |d} ||z }|dz }||d| dzz }||d| dzz }||d|j dzz }||d|j dzz }|S)EFor a given (pre-calculated) `OperatorStatsSummary` object (e.g. generated from `OperatorStatsSummary.from_block_metadata()`), returns a human-friendly string that summarizes operator execution statistics. Returns: String with summary statistics for executing the given operator. rrz6* Remote wall time: {} min, {} max, {} mean, {} total rUrXr7rz5* Remote cpu time: {} min, {} max, {} mean, {} total z.* UDF time: {} min, {} max, {} mean, {} total z8* Peak heap memory usage (MiB): {} min, {} max, {} mean z?* Output num rows per block: {} min, {} max, {} mean, {} total zA* Output size bytes per block: {} min, {} max, {} mean, {} total z?* Output rows per task: {} min, {} max, {} mean, {} tasks used r8z9* Tasks per node: {} min, {} max, {} mean; {} nodes used rz* Operator throughput: z * Total input num rows: z rows z * Total output num rows: rrz% * Estimated single task throughput: )rr%rrr4rr&rrr'r)r(rr,r) rGrrr^r_rar`rcrdr)node_count_statstotal_num_in_rowsrs r3__str__zOperatorStatsSummary.__str__ s -52..   6MC LSSOE*++OE*++OF+,,OE*++  CM   6MC KRRIe$%%Ie$%%If%&&Ie$%%  CM   6MC DKKIe$%%Ie$%%If%&&Ie$%%  C{   6MC NUUU#U#V$ C !% 4  6MC Rf%e,%e,%f-%e,  C#'"8 "  6MC Tf'.'.'/'.  CN   6MC Rf% % &!'"  C ?   6MC OVV ' ' ( )  C   4#; -1-FM))A "7u!=  6MC - -C V9JVVVV C X:LXXXX C ' C , C  r5rcrt|}||jrtdndz }d|jpiD}d|jpiD}d|jpiD}d|jpiD}d|jpiD}d|jpiD}d g|d |d |j d |d |jd |dt|j d |d|j |d|pdd |d|pdd |d|pdd |d|pdd |d|pdd |d|pdd |d} | S)rir+rc4i|]\}}|t|Sr;rrrsrts r3r1z1OperatorStatsSummary.__repr__..s$PPPA1c!ffPPPr5c4i|]\}}|t|Sr;rros r3r1z1OperatorStatsSummary.__repr__..s$III41aQAIIIr5c4i|]\}}|t|Sr;rros r3r1z1OperatorStatsSummary.__repr__..s$JJJda3q66JJJr5c4i|]\}}|t|Sr;rros r3r1z1OperatorStatsSummary.__repr__..s1! ! ! !QAs1vv! ! ! r5c4i|]\}}|t|Sr;rros r3r1z1OperatorStatsSummary.__repr__..s1# # # !QAs1vv# # # r5c4i|]\}}|t|Sr;rros r3r1z1OperatorStatsSummary.__repr__..s$RRR$!QAs1vvRRRr5zOperatorStatsSummary( z operator_name='z', z is_suboperator=rz time_total_s=z block_execution_summary_str=z wall_time=Nz cpu_time=z memory=z output_num_rows=z output_size_bytes=z node_count=r)r<rrrmrrrr'r(rrbr4rpr%) rGrrr^r_r`rcrdnode_conut_statsrs r3rzOperatorStatsSummary.__repr__s &&t';C.###CPP$.2FB1M1M1O1OPPPIIDM,?R+F+F+H+HIII JJt{/@b.G.G.I.IJJJ ! ! #'#7#=2"D"D"F"F! ! ! # # #'#9#?R"F"F"H"H# # # SR4?3Hb2O2O2Q2QRRR    v        )-);        )-)=        (+4+<'='=          7;6V        %4$;t        $-#4        ".!5        +@*G4        -D,Kt        &6%=            r5r )r\r]r^r1r"r#rBrr rrr&rrrrr'r(r)rrr, classmethodr rrrlrr;r5r3rrEsP !$$$$-1IxS%Z()000+/HhtCJ'(///+/HhtCJ'(///)-FHT#u*% &---*.(3-...26OXd3:./66648xS%Z 01888-1Jc5j)*111,0IxS%Z()000 ????X?CUCCCXCR R *%R  R R R R [R hvvvvvp$$3$$$$$$r5rceZdZUeed<eed<eed<eed<eed<eed<eed<eed<eed <eed <eed <eed <eed <eed<eed<eed<eed<defdZdefdZddefdZ dS)r wait_timeget_ref_bundles_timeget_time next_time format_time collate_timefinalize_batch_timetime_to_first_batch block_time user_timeinitialize_time total_timestreaming_split_coord_timerrrrr)c*|SrJ)rrFs r3rlzIterStatsSummary.__str__s~~r5c <d}|js|js|js|js~|jse|jsL|js3|js|j r7|dz }|jr<|d t|jz }|j r<|d t|j z }|jr<|d t|jz }|jr<|d t|jz }|j r<|d t|j z }|dz }|jr|d t|jt|jt|jt|jz }|jr|d t|jt|jt|jt|jz }|jrd }|| t|jt|jt|jt|jz }|jrd }|| t|jt|jt|jt|jz }|jr|d t|jt|jt|jt|jz }|j rd}|| t|j t|j t|j t|j z }t#jjr\|dz }|d |jz }|d |jz }|d |jz }|jr|d |jz }|jdkr1|dz }|t|jdz }|S)Nrz" Dataset iterator time breakdown: z* Total time overall: {} z> * Total time in Ray Data iterator initialization code: {} zE * Total time user thread is blocked by Ray Data iter_batches: {} zP * Total time spent waiting for the first batch after starting iteration: {} z/ * Total execution time for user thread: {} zC* Batch iteration time breakdown (summed across prefetch threads): z: * In get RefBundles: {} min, {} max, {} avg, {} total z5 * In ray.get(): {} min, {} max, {} avg, {} total z: * In batch creation: {} min, {} max, {} avg, {} total z< * In batch formatting: {} min, {} max, {} avg, {} total z6 * In collate_fn: {} min, {} max, {} avg, {} total zA * In host->device transfer: {} min, {} max, {} avg, {} total zBlock locations: z * Num blocks local: {} z * Num blocks remote: {} z& * Num blocks unknown location: {} z * Prefetched bytes: {} rz+Streaming split coordinator overhead time: r)rrSrrryrzr{r|r}r~rr4rrrUrXrZrrw'enable_get_object_locations_for_metricsrrrrr)rGrbatch_creation_str format_strs r3rzIterStatsSummary.to_stringsB O   ! !b I'++--b I""$$b I(,,.. b I }  "" b I ~!!## b I##%%b I $$&&b I'++--b I 9 9C""$$ W3::3t?R?R?T?T;U;UVVV#'')) !6#d&:&>&>&@&@"A"ABB""$$ !6#do&9&9&;&;"<"<=='++-- !6#d&>&B&B&D&D"E"EFF~!!## IPP**,,-- V C(,,.. T[[1557788155778815577881557788  }  "" OVV ))++,, ))++,, ))++,, ))++,,  ~!!## Q#)00**,,--**,,--**,,--**,,--  ##%% Sz(((,,..//(,,..//(,,..//(,,..//   $$&& PWW)--//00)--//00)--//00)--//00  '++-- Xz((0446677044667704466770446677  &((P ++5<UVVV@GG.) Y5<*>*@*@&A&A&IT        %((:(:(<(<$=$=$E        &))<)<)>)>%?%?%G4           r5Nr ) r\r]r^r>r"rr1rlrrr;r5r3rrs=OOO %%%%     f3ffffP  3      r5r)rr6)OrrRloggingrKr contextlibr dataclassesrrtypingrrr r r r r ruuidrnumpyrYr! ray.actorrray.data._internal.block_listr*ray.data._internal.execution.dataset_stater.ray.data._internal.execution.interfaces.commonr:ray.data._internal.execution.interfaces.op_runtime_metricsrrrrr$ray.data._internal.metadata_exporterrrrray.data._internal.utilrray.data.blockrray.data.contextrray.util.annotationsrray.util.metricsr r!r"r#ray.util.scheduling_strategiesr$ getLoggerr\rhrmrnr<r1r|rBr4rr<r>rarorrqrsrdrrrr;r5r3rs  ######%%%%%%))))))))                     !!!!!!333333CCCCCCRRRRRR -,,,,,%%%%%%((((((------>>>>>>>>>>>>IIIIII  8 $ $).  d:&& ' ;;3;;;;+++C+++++#V#V#V#V#V#V#V#VL********Z Ql l l l l l l l ^;{#;