o 1 i@@s>UddlZddlZddlZddlZddlZddlmZddlmZddlm Z m Z ddl m Z m Z mZmZmZmZmZmZddlmZddlZddlZddlmZddlmZdd lmZdd lm Z m!Z!m"Z"m#Z#m$Z$dd l%m&Z&m'Z'm(Z(dd l)m*Z*dd l+m,Z,ddl-m.Z.ddl/m0Z0ddl1m2Z2m3Z3m4Z4m5Z5ddl6m7Z7e8e9Z:dZ;dZee,fZ?de@de>fddZAd2deBdeBde>fddZCGdddZDGd d!d!ZEejFdd"Gd#d$d$ZGeHZIejHeJd%<d&d'ZKGd(d)d)ZLeLZMGd*d+d+ZNe0e Gd,d-d-ZOe Gd.d/d/ZPe Gd0d1d1ZQdS)3N) defaultdict)contextmanager) dataclassfields)AnyDictListMappingOptionalSetTupleUnion)uuid4) ActorHandle) BlockList) DatasetState) NODE_UNKNOWN MetricsGroup MetricsType NodeMetricsOpRuntimeMetrics)DatasetMetadataTopologyget_dataset_metadata_exporter)capfirst) BlockStats) DataContext) DeveloperAPI)CounterGauge HistogramMetric)NodeAffinitySchedulingStrategyZdatasets_stats_actorZ_dataset_stats_actorunknownsecondsreturncCsR|dkr tt|ddS|dkrtt|dddStt|ddddS)NsgMbP?msus)strround)r$r.d/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/data/_internal/stats.pyfmt/s r0lvlspaces_per_indentcCs d||S)zReturns a string of spaces which contains `level` indents, each indent containing `spaces_per_indent` spaces. For example: >>> leveled_indent(2, 3) ' '  r.)r2r3r.r.r/leveled_indent8 r5c@speZdZdZddZedddZdeddfd d Zdefd d Z defd dZ defddZ defddZ dS)Timerz8Helper class for tracking accumulated time (in seconds).cCs d|_td|_d|_d|_dS)Nrinf)_totalfloat_min_max _total_countselfr.r.r/__init__Ds  zTimer.__init__r%Nc cs>t}zdVW|t|dS|t|wN)time perf_counteradd)r?Z time_startr.r.r/timerJs *z Timer.timervaluecCs@|j|7_||jkr||_||jkr||_|jd7_dSNr&)r9r;r<r=)r?rFr.r.r/rDRs   z Timer.addcC|jSrA)r9r>r.r.r/getZz Timer.getcCrHrA)r;r>r.r.r/min]rJz Timer.mincCrHrA)r<r>r.r.r/max`rJz Timer.maxcCs|jr |j|jStdS)Nr8)r=r9r:r>r.r.r/avgcsz Timer.avgr%N) __name__ __module__ __qualname____doc__r@rrEr:rDrIrKrLrMr.r.r.r/r7As r7c@sNeZdZdZdedddeefddZded dfd d Z d e d dfd dZ dS)_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_timecCs||_||_|p t|_dSrA)rTrUrBrC start_time)r?rTrUrWr.r.r/r@nsz_DatasetStatsBuilder.__init__metadatar%cCsi}t|D].\}\}}t|}t|dkr1|dkr$|||j|<q|||jdd|<q|||j<qt||j|jd}t |j |_ |S)Nr&rz->)rYrU base_name) enumerateitemsrlenrTsplitrVrUrBrCrX time_total_s)r?rYZ op_metadataikvZcapped_kstatsr.r.r/build_multioperatorxs  z(_DatasetStatsBuilder.build_multioperator final_blockscCs,t|j|i|jd}t|j|_|S)N)rYrU)rVrT get_metadatarUrBrCrXr`)r?rfrdr.r.r/builds  z_DatasetStatsBuilder.buildN) rOrPrQrRr,r r:r@ StatsDictrerrhr.r.r.r/rSgs  rS)Znum_cpusc@s|eZdZdZd3ddZdedeedfdeee ffd d Z deee ffd d Z d dZ ddZ d4dedeeeeeeffdeedeeefdeeeeeeeeffff ddZd5ddZ  d6ddZdededeed ed!ef d"d#Zdedeeeffd$d%Zd4deefd&d'Zd(ed)efd*d+Zd(ed,eeeffd-d.Z  d7ded/eed0eefd1d2ZdS)8 _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)cCstt|_i|_i|_||_d|_i|_i|_ t |_ i|_ t |_d}tdd|d|_tdd|d|_tdd |d|_td d |d|_td d |d|_tdd|d|_tdd|d|_|jtj|d|_|jtj|d|_|jtj|d|_|jtj|d|_ |jtj!|d|_"|jtj#|d|_$|%|_&d}tdd|d|_'tdd|d|_(tdd|d|_)tdd|d|_*tdd|d|_+tdd|d|_,tdd |d|_-td!d"|d|_.td#d$|d|_/td%d&|d|_0td'd(|d|_1td)d*|d|_2td+d,|d|_3td-d.|d|_4td/d0|d|_5td1d2|d|_6td3d4|d|_7d5}td6d7|d|_8td8d9|d|_9td:d;d<:d=d>t;Dd?|d|_tdDdE|d|_?tdFdGd<:dHd>t;Dd?|d|_@dS)INrdatasetoperatorZdata_spilled_byteszBytes spilled by dataset operators. DataContext.enable_get_object_locations_for_metrics must be set to True to report this metric descriptiontag_keysZdata_freed_bytesz Bytes freed by dataset operatorsZdata_current_bytesz9Bytes currently in memory store used by dataset operatorsZdata_cpu_usage_coresz#CPUs allocated to dataset operatorsZdata_gpu_usage_coresz#GPUs allocated to dataset operatorsZdata_output_bytesz$Bytes outputted by dataset operatorsZdata_output_rowsz#Rows outputted by dataset operators) metrics_grouprp)rlZ%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.Z data_iter_block_fetching_secondsz>Seconds taken to fetch (with ray.get) blocks by iter_batches()Zdata_iter_batch_shaping_secondszCSeconds taken to shape batch from incoming blocks by iter_batches()Z"data_iter_batch_formatting_secondsz1Seconds taken to format batches by iter_batches()Z!data_iter_batch_collating_secondsz2Seconds taken to collate batches by iter_batches()Z"data_iter_batch_finalizing_secondsZdata_iter_total_blocked_secondsz0Seconds user thread is blocked by iter_batches()Zdata_iter_user_secondszSeconds spent in user codeZdata_iter_initialize_secondsz-Seconds spent in iterator initialization codeZdata_iter_get_secondsz;Seconds spent in ray.get() while resolving block referencesZdata_iter_next_batch_secondsz:Seconds spent getting the next batch from the block bufferZdata_iter_format_batch_secondsz"Seconds spent formatting the batchZdata_iter_collate_batch_secondsz!Seconds spent collating the batchZ data_iter_finalize_batch_secondsz"Seconds spent finalizing the batchZdata_iter_blocks_localz*Number of blocks already on the local nodeZdata_iter_blocks_remotez8Number of blocks that require fetching from another nodeZdata_iter_unknown_locationz,Number of blocks that have unknown locationsrljob_idrX#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, cSg|] }|jd|jqS=rFname.0r(r.r.r/ z(_StatsActor.__init__..)$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 (cSrwrxrzr|r.r.r/r~r)A collectionsrdictrYZ last_timerX 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_metricsrZINPUTSexecution_metrics_inputsZOUTPUTSexecution_metrics_outputsZTASKSexecution_metrics_tasksZOBJECT_STORE_MEMORY"execution_metrics_obj_store_memoryZACTORSexecution_metrics_actorsZMISCexecution_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_s iter_get_siter_next_batch_siter_format_batch_siter_collate_batch_siter_finalize_batch_siter_blocks_localiter_blocks_remoteiter_unknown_locationrtrujoinrrvrrrr)r?rZ op_tags_keysZ iter_tag_keys dataset_tags operator_tagsr.r.r/r@s          z_StatsActor.__init__rqrp.r%cCsi}tD]H}|j|ksqd|j}|j}|jtjkr(t|||d||j<q|jtjkr>t|f||d|j ||j<q|jtj krNt |||d||j<q|S)Ndata_rn) rZ get_metricsrqr{roZ metrics_typerrr Z metrics_argsr)r?rqrpmetricsmetric metric_nameZmetric_descriptionr.r.r/rs:       z<_StatsActor._create_prometheus_metrics_for_execution_metricscCs8i}ttD]}d|jd}t|ddd||j<q|S)NrZ _per_node)rlnode_iprn)rrr{r)r?rfieldrr.r.r/rs z;_StatsActor._create_prometheus_metrics_for_per_node_metricscCst|j}|jd7_|SrG)r,r)r? dataset_idr.r.r/get_dataset_ids z_StatsActor.get_dataset_idcCs,|D]}|j|q|D]}|j|q dSrA)update_execution_metricsupdate_iteration_metrics)r?execution_metricsiteration_metricsrr.r.r/update_metricss   z_StatsActor.update_metricsN dataset_tag op_metricsrstaterc Cs` ddtdtttttfdtttffdd}t||D]\}}|||} |j | dd| |j | dd| |j | d d| |j | d d| |j | d d| |j | d d| |j | d d| |jD]\} } || | | d| qy|jD]\} } || | | d| q|jD]\} } || | | d| q|jD]\} } || | | d| q|jD]\} } || | | d| q|jD]\} } || | | d| qq|dur(|D]1\} } | |jvr||j | t}|j||d} | D]\}}|j|} || || qq|||dS)N prom_metricrFtagscSst|tr |||dSt|tr|||dSt|trct|trett|D]<}|t|dkr9|j |n|j dd}|dkrK|j |dnd}||d}t||D]}| ||qYq*dSdSdS)Nr&rZdrr') isinstancersetrincr listranger^Z boundariesZobserve)rrFrraZboundary_upper_boundZboundary_lower_boundZ bucket_value_r.r.r/_records(      z5_StatsActor.update_execution_metrics.._recordZobj_store_mem_spilledrZobj_store_mem_freedZobj_store_mem_usedZbytes_task_outputs_generatedZrow_outputs_takenZ cpu_usageZ gpu_usage)r node_ip_tagrA)r!r intr:rrr,zip _create_tagsrrrIrrrrrrrr]rrrrrr_rebuild_ray_nodes_cacherrupdate_dataset)r?rrrrrrrd operator_tagr field_namernode_id node_metricsrrZ metric_valuer.r.r/rsX       z$_StatsActor.update_execution_metricscCsHt}|D]}|dd}|dd}|dur!|dur!||j|<qdS)NZNodeIDZNodeName)raynodesrIr)r?Z current_nodesnoderZ node_namer.r.r/r(s   z$_StatsActor._rebuild_ray_nodes_cacherdrVcCsV||}|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|dSrA)rrrrIrrrrrrrrrrrrrriter_time_to_first_batch_srr)r?rdrrr.r.r/r0s$ z$_StatsActor.update_iteration_metricsrstopology data_contextc Csxt}|tjjddd|ddd|Dd|j|<|jdur:t|||||ddtjjd|j|<|j|j|dSdS)NrcSs i|] }|tjjddddqS)r)rprogresstotal queued_blocks)rPENDINGr{)r}rmr.r.r/ ]sz0_StatsActor.register_dataset..)rsrrr total_rowsrXend_time operators)rsrrrXrexecution_start_timeexecution_end_timer) rBrrr{rrrrexport_dataset_metadata)r?rsrrrrrXr.r.r/register_datasetLs6   z_StatsActor.register_datasetc Cs|j|||j|}|j|dd}tt|j|dd}|||d}|j|dd||j|dd||dtj j }t |}|j |j ||||i}|d iD]C\} } || d } |j| dd| |j| dd| |j| d d| | dtj j }t |}|j|j | ||| <qd||||dtjj tjj hvr|j|t|j|jkr|jr|j} |j| d|j| dt|j|jkr|jsdSdSdSdSdS) NrsNonerXrrrrrrrrkr)rupdaterIr,rrtrrurUNKNOWNr{Z from_stringrvrFupdate_dataset_metadata_stater]rrrr'update_dataset_metadata_operator_statesFINISHEDFAILEDrappendr^rpopleftpopr) r?rrrsrXrZ state_stringZ state_enumoperator_statesrmZop_staterZ tag_to_evictr.r.r/rvs\              z_StatsActor.update_datasetcs"s|jSfdd|jDS)Ncs"i|] \}}|dkr||qSrsr.r}rbrcrr.r/rs"z,_StatsActor.get_datasets..)rr])r?rsr.rr/ get_datasetssz_StatsActor.get_datasetsr new_statecCs||jvrdSt}|j|}|j|krdSt|}||_|tjjkr)||_n!|tj jtj jfvrJ||_ |j j D]}|jtjjkrI||_||_ q:||j|<|j|dSrA)rrBrcopydeepcopyrRUNNINGr{rrrrrrrr)r?rr update_timedataset_metadataupdated_dataset_metadatarmr.r.r/rs&       z)_StatsActor.update_dataset_metadata_staterc Cs||jvrdS|j|}d}|jjD]}|j|vr%|j||jkr%d}nq|s*dSt|}t}|jjD]2}|j|vri||j}|j|krIq7||_|tj j krV||_ q7|tj j tj j fvri||_|j si||_ q7||j|<|j|dS)NFT)rrridrrrrBrrr{rrrrrr) r?rrrZ update_neededrmrrrr.r.r/rs@           z3_StatsActor.update_dataset_metadata_operator_statesrrcCs,d|i}|dur ||d<|dur||d<|S)Nrlrmrr.)r?rrrrr.r.r/rs z_StatsActor._create_tags)r)rArN)rdrV)NN) rOrPrQrRr@rr r,rr!rrrrrrr rr:rr rrrrrrrrrrrr.r.r.r/rjst |       Y  *9  -rj_stats_actor_lockcCspt}|j}tjjjstt dd}t t j t tdd|dWdS1s1wYdS)NF)ZsoftTZdetached)r{ namespaceZ get_if_existsZlifetimescheduling_strategy)r get_currentrrutilclientZ is_connectedr"get_runtime_contextZ get_node_idrrjoptionsSTATS_ACTOR_NAMESTATS_ACTOR_NAMESPACEremote)ctxrr.r.r/_get_or_create_stats_actors" $r c @seZdZdZdZdZddZ d$dedee fdd Z d d Z d e e deeeeeeeeffffd dZ d$ded e e de edeeefdef ddZdefddZdddefddZdefddZdede ededefdd Zdefd!d"Zd#S)% _StatsManagerafA Class containing util functions that manage remote calls to _StatsActor. This class collects stats from execution and iteration codepaths and keeps track of the latest snapshot. An instance of this class runs a single background thread that periodically forwards the latest execution/iteration stats to the _StatsActor. This thread will terminate itself after being inactive (meaning that there are no active executors or iterators) for STATS_ACTOR_UPDATE_THREAD_INACTIVITY_LIMIT iterations. After terminating, a new thread will start if more calls are made to this class. cCs6d|_d|_i|_i|_t|_d|_t|_dSrA) _stats_actor_handle_stats_actor_cluster_id_last_execution_stats_last_iteration_stats threadingLock _stats_lock_update_thread_update_thread_lockr>r.r.r/r@1s z_StatsManager.__init__F skip_cacher%cCstjjjdur tdtjjjj}|jdus|j|ks|rCztjt t d|_||_W|jSt yBt |_tjjjj|_Y|jSw|jS)NzEGlobal node is not initialized. Driver might be not connected to Ray.)r{r) rZ_privateZworkerZ _global_node RuntimeErrorZ cluster_idrrZ get_actorrr  ValueErrorr )r?rZcurrent_cluster_idr.r.r/r Cs,     z(_StatsManager._get_or_create_stats_actorcsvj.jdusjs)fdd}tj|dd_jWddSWddS1s4wYdS)Nc sd} js jrmzR}|durWqg}j(jD]\}}}}}dd|D}|||||f} || qWdn1sDwY|jjt|tjdd}Wn!t ylt j dddYdSw|d7}|t j kr}t d dSttjq) NrTcSg|]}|jddqST)Zreset_histogram_metricsas_dictr}rr.r.r/r~{s zX_StatsManager._start_thread_if_not_running.._run_update_loop..)rrz1Error occurred during remote call to _StatsActor.)exc_infor&z2Terminating StatsManager thread due to inactivity.)rrr rvaluesrrr r Exceptionloggerdebugr UPDATE_THREAD_INACTIVITY_LIMITrBsleep StatsManager#STATS_ACTOR_UPDATE_INTERVAL_SECONDS) Ziter_stats_inactivity stats_actorZformatted_execution_statsrrrrrop_metrics_dictsargsr>r.r/_run_update_loopesh    zD_StatsManager._start_thread_if_not_running.._run_update_loopT)targetdaemon)rris_aliverThreadstart)r?r,r.r>r/_start_thread_if_not_running`s < "z*_StatsManager._start_thread_if_not_runningrc CsltjsdStdd}|D]$}|jD]\}}||}ttD]}||jt ||j7<q"qq|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. NcSsttSrA)rrr.r.r.r/sz;_StatsManager._aggregate_per_node_metrics..) rrZenable_per_node_metricsrZ_per_node_metricsr]rrr{getattr)r?rZaggregated_by_noderrrZagg_node_metricsfr.r.r/_aggregate_per_node_metricss   z)_StatsManager._aggregate_per_node_metricsrrr force_updatec Cs||}|rdd|D}|||||f}|jj|dS|||||f}|j ||j|<Wdn1s9wY|dS)NcSrrrrr.r.r/r~s z:_StatsManager.update_execution_metrics..)r6r rr rrr2) r?rrrrr7rr*r+r.r.r/rs"   z&_StatsManager.update_execution_metricscCN|j||jvr|j|=WddSWddS1s wYdSrA)rrr?rr.r.r/clear_last_execution_statss   "z(_StatsManager.clear_last_execution_statsrdrVcCs@|j||f|j|<Wdn1swY|dSrA)rrr2)r?rdrr.r.r/rs z&_StatsManager.update_iteration_metricscCr8rA)rrr9r.r.r/clear_iteration_metricss   "z%_StatsManager.clear_iteration_metricsrrcCs,|jdd}|jt||||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 TrN)r rr rrZ get_job_id)r?rrrrr)r.r.r/register_dataset_to_stats_actors  z-_StatsManager.register_dataset_to_stats_actorcCs:z|jdd}t|jWStytjYSw)NTr<)r rrIrr r"rhex)r?r)r.r.r/get_dataset_id_from_stats_actors    z-_StatsManager.get_dataset_id_from_stats_actorN)F)rOrPrQrRr(r%r@boolr rr r2rrr r,r rr:r6rrrr:rr;rrr=r?r.r.r.r/r sT H      r c@s~eZdZdZdddedeededfdefddZ e d d Z dd ed ee d e fddZdddZd efddZdS)rVzHolds 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)r[rYrUr[cCs||_|durt|ts|g}|pg|_|jsdn tdd|jDd|_||_d|_d|_t |_ t |_ t |_ t |_ t |_t |_t |_t |_t |_t |_t |_t |_i|_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. Nrcs|]}|jVqdSrA)numberr}pr.r.r/ @z(DatasetStats.__init__..r&Z unknown_uuid)rYrrparentsrLrBr[ dataset_uuidr`r7streaming_exec_schedule_s iter_wait_srrrrrrrrr iter_total_s extra_metricsrrrglobal_bytes_spilledglobal_bytes_restoreddataset_bytes_spilledstreaming_split_coordinator_s)r?rYrUr[r.r.r/r@*s:   zDatasetStats.__init__cCstSrA)r r>r.r.r/r)kszDatasetStats.stats_actorr{rWr%cCs t|||S)z>Start recording stats for an op of the given name (e.g., map).)rS)r?r{rWr.r.r/ child_builderos zDatasetStats.child_builderDatasetStatsSummaryc sg}t|jdkt|j|j|j|j|j|j|j |j |j |j |j |j|j|j|j}g}|jdur9dd|jD}d}t|D]1\}}|jrp|jd}t|jtrX|jddnd}td|dd |jd |d ||7}q?fd d|jD} t| D]-\}} r|dkr|| _n| |d} | jrd| jvr| jdnd| _n|| _|| q|jr|jnd} t ||||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&NcSg|]}|qSr.) to_summaryrCr.r.r/r~z+DatasetStats.to_summary..rrZsumzParent z (operator: z) contributes z rows to inputcs g|] \}}tj||dqS)is_sub_operator)OperatorStatsSummaryfrom_block_metadata)r}r{rdrWr.r/r~s ))r^rYIterStatsSummaryrJrrrrrrrrrrKrPrrrrGr\operators_statsroutput_num_rowsrrIr#r$rTr]total_input_num_rowsrrIrRrBrHr`r[rLrMrNrO) r?r\ iter_statsZstats_summary_parentsZparent_total_outputraZparent_summaryZlast_parent_opZ op_outputZop_statsZop_statZprev_oprIr.rWr/rTus         zDatasetStats.to_summarycCs |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.)rTruntime_metricsr>r.r.r/r`r6zDatasetStats.runtime_metricsrA)r%rR)rOrPrQrRrir r rr,r@propertyr)r:rSrQrTr`r.r.r.r/rV#s,  A  ]rVc@s2eZdZUeded<ded<eded<eed<eed<eed<eed <eee fed <eed <eed <eed <eed<   d)de e ede defddZ edddedfddZedddeeeffddZdefddZd*defdd Zdefd!d"Zdefd#d$Zdefd%d&Zdefd'd(ZdS)+rRrYr\r[r_rGrBrHr`r[rLrMrNrOrINTalready_printedinclude_parentr%cCs|durt}d}|jr%|r%|jD]}|j|dd}|r$||7}|d7}qd}t|jdkrY|jd}|j}|j|} |d|j|7}| |vrM|d 7}n\| | |t |7}nPt|jdkrt |j d } | dkrld} |d |j|j | 7}t|jD],\} }|j}|j|} |d7}|d | |7}| |vr|d 7}q|| | |t |7}q|tj} | r|jr|r|jrdnd} || 7}|dt |jd7}|t |j7}t|jdkrS|rSt |jd}t |jd}|s|r|d7}|d|7}|d|7}t |jd}|r|d7}|d|7}|jdj}|r$|dnd}|}|}|rS|rS|rS|d7}|d7}|d||d7}|d||d7}| ra|ra|d|7}|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: rZrVzDataset throughput:  * Ray Data throughput:  rows/s % * Estimated single node throughput: )rrG to_stringr^r\rTrHformatrBrDr,r-r`r[r\rrverbose_stats_logsrLrXr_rMrNrOr]get_total_wall_timeget_total_time_all_blocksr`)r?rbrcrdoutrD parent_sumZoperators_stats_summaryrTZ operator_uuid rounded_totalnrmindentZ mb_spilledZ mb_restoredZdataset_mb_spilledr]total_num_out_rows wall_timeZtotal_time_all_blocksr.r.r/rks                zDatasetStatsSummary.to_stringcurrcCs4g}|jD]}|r|jr|t|q||gSrA)rGextendrR _collect_dataset_stats_summaries)rwZsummsrDr.r.r/ry_s    z4DatasetStatsSummary._collect_dataset_stats_summariessummcCs0tdd|jD}tdd|jD}||fS)NcsrArA)earliest_start_timer}opsr.r.r/rElrFz:DatasetStatsSummary._find_start_and_end..csrArA)latest_end_timer|r.r.r/rEmrF)rKr\rL)rzearliest_start latest_endr.r.r/_find_start_and_endjsz'DatasetStatsSummary._find_start_and_endcs|dtdtdtffdd }t|}d}|D]}t|jdkr6t|\}}||}|||j|7}q||d|j 7}||d 7}|S) Nr{rBr%cs6dkr|nd}d|dt|d|dddS)Nrz* : z (rz.3fz%) r0)r{rBfractionZtotal_wall_timer.r/fmt_liness"z5DatasetStatsSummary.runtime_metrics..fmt_linezRuntime Metrics: rZ SchedulingZTotal) rnr,r:rRryr^r\rr[rI)r?r summariesrprzrrZ op_total_timer.rr/r`ps z#DatasetStatsSummary.runtime_metricsrcst}dfdd|jD}dfdd|jD}dfdd|jD}|r8d|d|dnd }|rEd|d|dnd }|rRd|d|dnd }d g|d |d |jd|d |jd|d |jd|d|d|d|d|d|j dd|d|j dd|d|j dd|d|j dd|d|d|dS)Nrecg|] }|dqSr'__repr__r}sslevelr.r/r~z0DatasetStatsSummary.__repr__..crrr)r}Zpsrr.r/r~rc3s0|]\}}td|d|dVqdS)r'r,N)r5rrr.r/rEs  z/DatasetStatsSummary.__repr__.., 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=rgzMB, z global_bytes_restored=z dataset_bytes_spilled=z parents=[r)r5rr\rGrLr]rHr[rBr_rrMrNrO)r?rrtr\Z parent_statsrLr.rr/rsp     zDatasetStatsSummary.__repr__cCsPddt|D}t|dkrdStdd|D}tdd|D}||S)zCalculate the total wall time for the dataset, this is done by finding the earliest start time and latest end time for any block in any operator. The wall time is the difference of these two times. cSs$g|]}t|jdkrt|qSr)r^r\rRrr}rzr.r.r/r~s z;DatasetStatsSummary.get_total_wall_time..rcs|]}|dVqdS)rNr.r}Z start_endr.r.r/rEz:DatasetStatsSummary.get_total_wall_time..csr)r&Nr.rr.r.r/rEr)rRryr^rKrL)r?Z start_endsrrr.r.r/rns z'DatasetStatsSummary.get_total_wall_timecCst|}tdd|DS)zGCalculate the sum of the wall times across all blocks of all operators.css$|] }tdd|jDVqdS)css(|]}|jr|jddndVqdSrVrN)rvrIr|r.r.r/rEs  zJDatasetStatsSummary.get_total_time_all_blocks...N)rVr\rr.r.r/rEs z@DatasetStatsSummary.get_total_time_all_blocks..)rRryrV)r?rr.r.r/ros z-DatasetStatsSummary.get_total_time_all_blockscCs,tdd|jD}|tdd|jDS)Ncss|]}|VqdSrA)get_total_cpu_timerCr.r.r/rErz9DatasetStatsSummary.get_total_cpu_time..css|] }|jddVqdSr)cpu_timerIrr.r.r/rEs )rVrGr\)r?rqr.r.r/rs  z&DatasetStatsSummary.get_total_cpu_timecCsFdd|jD}|rt|nd}|js|St|gdd|jDRS)NcSrSr.)get_max_heap_memoryrCr.r.r/r~rUz;DatasetStatsSummary.get_max_heap_memory..rcSsg|] }|jddqS)rLr)memoryrIrr.r.r/r~r)rGrLr\)r?Z parent_memoryZ parent_maxr.r.r/rsz'DatasetStatsSummary.get_max_heap_memory)NTTr)rOrPrQr__annotations__rr,r:rrr r r@rk staticmethodryr rr`rrnrorrr.r.r.r/rRsJ    t    rRc @sNeZdZUeed<eed<eed<eed<eed<eed<dZee eefed<dZ ee eefed <dZ ee eefed <dZ ee eefed <dZ eeed <dZee eefed <dZee eefed<dZee eefed<dZee eefed<ededeededdfddZdefddZddefddZdS)rYrTrXr`r{r~block_execution_summary_strNrvrudf_timerr^r]output_size_bytes node_count task_rows block_statsr%cCs\dd|D}d}d}d\}}|r'tdd|D}tdd|D}||}|r1dt|} n|rGt|d }|dkr>d}d t||} nd } | d 7} tt} |D]} | jd urk| j d urk| | j j | j7<qTd } t| dkrt| t| tt t| t| d} dt| | } d\} }}}|r!tdd|Dtdd|Dt dd|Dtdd|Dd} tdd|Dtdd|Dt dd|Dtdd|Dd}dd|D}t|t|tt |d}tdd|Dtdd|Dt dd|Dtdd|Dd}d }d d|D}|r?t|t|tt |t|d}d }d!d|D}|r]t|t|tt |t|d}d }|rtt}|D] }||j|j qid"d#|D}t| t| tt t| t|d}d }t|||||| | |||||||| d$S)%aCalculate the stats for a operator from a given list of blocks, and generates a `OperatorStatsSummary` object with the results. Args: block_stats: List of `BlockStats` to calculate stats of operator_name: Name of operator associated with `blocks` is_sub_operator: Whether this set of blocks belongs to a sub operator. Returns: A `OperatorStatsSummary` object initialized with the calculated statistics cSg|] }|jdur|jqSrA) exec_statsr}mr.r.r/r~z.r)rrcsrArA)Z start_time_sr|r.r.r/rE rFz;OperatorStatsSummary.from_block_metadata..csrArA)Z end_time_sr|r.r.r/rErFz{} blocks produced r'z{} blocks produced in {}srreN)rKrLmeancountz{} tasks executed, {})NNNNcSg|]}|jqSr.Z wall_time_sr}er.r.r/r~3cSrr.rrr.r.r/r~4rcSrr.rrr.r.r/r~5rcSrr.rrr.r.r/r~6r)rKrLrrVcSrr.Z cpu_time_srr.r.r/r~9rcSrr.rrr.r.r/r~:rcSrr.rrr.r.r/r~;rcSrr.rrr.r.r/r~<rcSs g|] }t|jp dddqS)rir')r-Z max_uss_bytesrr.r.r/r~?s)rKrLrcSrr.Z udf_time_srr.r.r/r~IrcSrr.rrr.r.r/r~JrcSrr.rrr.r.r/r~KrcSrr.rrr.r.r/r~LrcSrrA)num_rowsrr.r.r/r~PrcSrrA)Z size_bytesrr.r.r/r~ZscSi|] \}}|t|qSr.)r^)r}rtasksr.r.r/rkrz.)rTrXr`r{r~rrvrrrr^r]rrr)rKrLrlr^r-rrrrrZtask_idxr!nprrrVrrrDr]rY)clsrTrrXrrrr`r{r~Zexec_summary_strrmetaZtask_rows_statswall_time_stats cpu_stats memory_stats udf_statsZmemory_stats_mboutput_num_rows_statsr]output_size_bytes_statsrZnode_counts_statsZ node_tasksr(Z node_countsr^r.r.r/rZs            z(OperatorStatsSummary.from_block_metadatac Cs|jrdnd}|j}|j}|r-||7}|dt|dt|dt|dt|d7}|j}|rP||7}|dt|dt|dt|dt|d7}|j}|rs||7}|d t|dt|dt|dt|d7}|j}|r||7}|d |d|d|d7}|j}|r||7}|d |d|d|d|d7}|j }|r||7}|d |d|d|d|d7}|j } | r||7}|d | d| d| d| d7}|j } | r||7}|d| d| d| d| d7}|rE|j rE|rE|j r|j nd} |d} ||7}|d7}||d| d7}||d| d7}||d| |j d7}||d| |dd7}|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. rfrz6* Remote wall time: {} min, {} max, {} mean, {} total rKrLrrVz5* 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 rz9* Tasks per node: {} min, {} max, {} mean; {} nodes used rz* Operator throughput: z * Total input num rows: z rows z * Total output num rows: rhrirj)rXrrvrlr0rrrr]rrrr`r^) r?rtrprrrrrrrZnode_count_statsZtotal_num_in_rowsrur.r.r/__str__s                 zOperatorStatsSummary.__str__rc Cst|}||jr tdnd7}dd|jpiD}dd|jp"iD}dd|jp.iD}dd|jp:iD}dd|jpFiD}d d|jpRiD}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)rr&rcSrr.rrr.r.r/rrz1OperatorStatsSummary.__repr__..cSrr.rrr.r.r/rrcSrr.rrr.r.r/rrcSrr.rrr.r.r/rcSrr.rrr.r.r/rrcSrr.rrr.r.r/rrzOperatorStatsSummary( 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)r5rXrvr]rrr]rrrrTr0r`r) r?rrtrrrrrZnode_conut_statsrpr.r.r/r sr         zOperatorStatsSummary.__repr__r)rOrPrQr,rr@r:rvr rrrrr^rr]rrr classmethodrrrZrrr.r.r.r/rYs<  rYc@seZdZUeed<eed<eed<eed<eed<eed<eed<eed<eed <eed <eed <eed <eed <eed<eed<defddZdefddZddefddZ dS)r[ wait_timeget_time next_time format_time collate_timefinalize_batch_timetime_to_first_batch block_time user_timeinitialize_time total_timestreaming_split_coord_timerrrr%cCs|SrA)rkr>r.r.r/rRszIterStatsSummary.__str__c Csd}|js+|js+|js+|js+|js+|js+|js+|jry|d7}|jr@|d t |j7}|j rQ|d t |j 7}|jrb|d t |j7}|jrs|d t |j7}|j r|d t |j 7}|d7}|jr|d t |j t |jt |jt |j7}|jrd }|| t |j t |jt |jt |j7}|jrd }|| t |j t |jt |jt |j7}|jr|d t |j t |jt |jt |j7}|jr?d }|| t |j t |jt |jt |j7}tjra|d7}|d |j7}|d |j7}|d |j7}|jdkry|d7}|t |jd7}|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): 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: {} rz+Streaming split coordinator overhead time: re)rrIrrrrrrrrlr0rrrKrLrMrrZ'enable_get_object_locations_for_metricsrrrr)r?rpZbatch_creation_str format_strr.r.r/rkUs                                zIterStatsSummary.to_stringrcCs:t|}dgd|dt|jpdd|dt|jp'dd|d|jp4dd|d|jpAdd|d|jpNdd|d t|j p_dd|d t|j ppdd|d t|j pdd|d t|j pdd|d S)NrzIterStatsSummary( z wait_time=rz get_time=z iter_blocks_local=z iter_blocks_remote=z iter_unknown_location=z next_time=z format_time=z user_time=z total_time=r) r5rr0rrIrrrrrrrr)r?rrtr.r.r/rsR    zIterStatsSummary.__repr__Nr) rOrPrQr7rrr,rrkrr.r.r.r/r[2s& ^r[)rr1)RrrloggingrrBr contextlibr dataclassesrrtypingrrrr r r r r uuidrnumpyrrZ ray.actorrZray.data._internal.block_listrZ*ray.data._internal.execution.dataset_staterZ:ray.data._internal.execution.interfaces.op_runtime_metricsrrrrrZ$ray.data._internal.metadata_exporterrrrZray.data._internal.utilrZray.data.blockrZray.data.contextrZray.util.annotationsrZray.util.metricsrrr r!Zray.util.scheduling_strategiesr" getLoggerrOr#rr rr,rir:r0rr5r7rSr rjRLockrrr r r'rVrRrYr[r.r.r.r/sp   (            & -s9x_