o `۷i!@sddlZddlmZddlmZmZmZmZmZm Z m Z m Z ddl Z ddl mZddlmZddlmZer$zBaseStatSpec.post_process_fnstat_fnr*cCs||_||_dSr$r+r*)selfr+r*rrr__init__ s zBaseStatSpec.__init__N)__name__ __module__ __qualname____doc__r r rr.rrrrr"s r"c sZeZdZdZddddddeeeegeffdedeed ee ffd d Z Z S) AggregateStatSpecz5Represents an AggregateFnV2 spec for a single column.cCr#r$rr%rrrr'1r(zAggregateStatSpec.N)r*column batch_format aggregator_fnr*r4r5cs tj||d||_||_dSNr,)superr.r4r5)r-r6r*r4r5 __class__rrr.-s  zAggregateStatSpec.__init__) r/r0r1r2r r rrr r r. __classcell__rrr9rr3*sr3c s`eZdZdZddddedeeegefdeeegefded eef fd d ZZ S) CallableStatSpeczPRepresents a user-defined stat function that operates outside Dataset.aggregate.cCr#r$rr%rrrr'Fr(zCallableStatSpec.r)r+ stat_key_fn post_key_fnr*columnscs&tj||d||_||_||_dSr7)r8r.r?r=r>r-r+r=r>r*r?r9rrr.@s  zCallableStatSpec.__init__) r/r0r1r2rr rrr.r;rrr9rr<=sr<c @seZdZdZddZddZdddd d eegefd ed e ed e e ddf ddZ dddddege fdeegefde eegefd ed e eddf ddZdddeee ffddZde efddZde efddZde efd d!Zd"d#Zd$d%ZdS)&StatComputationPlana\ Encapsulates a set of aggregators (AggregateFnV2) and legacy stat functions to compute statistics over a Ray dataset. Supports two types of aggregations: 1. AggregateFnV2-based aggregators, which are batch-executed using `Dataset.aggregate(...)`. 2. Callable-based stat functions, executed sequentially (legacy use case). cCs t|_dSr$)r _aggregatorsr-rrrr.\s zStatComputationPlan.__init__cCs|jdSr$)rBclearrCrrrreset_szStatComputationPlan.resetcCr#r$rr%rrrr'fr(zStatComputationPlan.N)r*r5r6r*r?r5rc Cs.|D]}||}|jt||||dqdS)a Registers an AggregateFnV2 factory for one or more columns. Args: aggregator_fn: A callable (typically a lambda or class) that accepts a column name and returns an instance of AggregateFnV2. The aggregator should set its name using alias_name parameter to control the output key. post_process_fn: Function to post-process the aggregated result. columns: List of column names to aggregate. batch_format: The batch format for aggregation results. If ARROW, results are kept in Arrow format for post_process_fn. Otherwise, results are converted to Python/pandas format. )r6r*r4r5N)rBappendr3)r-r6r*r?r5r4 agg_instancerrradd_aggregatorbsz"StatComputationPlan.add_aggregatorcCr#r$rr%rrrr'r()r>r*r+r=r>c Cs"|jt|||||p |ddS)ao Registers a custom stat function to be run sequentially. This supports legacy use cases where arbitrary callables are needed and cannot be run via Dataset.aggregate(). Args: stat_fn: A zero-argument callable that returns the stat. stat_key_fn: A callable that takes a column name and returns the key for the stat. post_key_fn: Optional; a callable to post-process the key. If not provided, stat_key_fn is used. post_process_fn: Function to post-process the result. columns: List of column names to compute the stat for. )r+r*r?r=r>N)rBrFr<r@rrradd_callable_statsz%StatComputationPlan.add_callable_statdatasetrc Csi}|}|rJ|dj|}|}|stdt|d}|D]$}|jj }| |d} |j t j kr@|| ||<q%|| ||<q%|D]!}||j} |jD]} || }|| } || ||| <qYqN|S)aq Executes all registered aggregators and stat functions. AggregateFnV2-based aggregators are batched and executed via Dataset.aggregate(). Callable-based stat functions are run sequentially. Args: dataset: The Ray Dataset to compute statistics on. Returns: A dictionary of computed statistics. NzAggregation returned no resultsr)_get_aggregate_fn_listgroupby aggregate to_arrow_refs ValueErrorrayget_get_aggregate_specsr+namer4r5r ARROWr*as_py_get_custom_stat_fn_specsr=r?r>) r-rJstats aggregatorsagg_ds arrow_refs arrow_tablespecstat_key agg_resultresultcolpost_keyrrrcomputes,        zStatComputationPlan.computecCdd|jDS)NcSsg|] }t|tr|jqSr) isinstancer3r+.0r\rrr s z>StatComputationPlan._get_aggregate_fn_list..rBrCrrrrKsz*StatComputationPlan._get_aggregate_fn_listcCrc)NcSg|] }t|tr|qSr)rdr3rerrrrg  z.rhrCrrrrRz(StatComputationPlan._get_aggregate_specscCrc)NcSrir)rdr<rerrrrgrjzAStatComputationPlan._get_custom_stat_fn_specs..rhrCrrrrVrkz-StatComputationPlan._get_custom_stat_fn_specscCst|dkS)Nr)lenrVrCrrrhas_custom_stat_fnsz&StatComputationPlan.has_custom_stat_fncCs t|S)z4 Iterates over all AggregatorSpecs. )iterrRrCrrr__iter__s zStatComputationPlan.__iter__)r/r0r1r2r.rErrr rr r rHrrIrrbrKr3rRr<rVrmrorrrrrARsL   $    - rA callbackscsfdd}|S)z Wraps a base post-processing function with a sequence of callback functions. Useful when multiple post-processing steps need to be applied in order. cs|}D]}||}q|Sr$r)r_ processedcbbase_fnrprrwrappers z$make_post_processor..wrapperr)rtrprurrsrmake_post_processorsrv)r collectionsrtypingrrrrrrr r rP"ray.air.util.data_batch_conversionr ray.data.aggregater ray.util.annotationsr ray.data.datasetrrrobjectrr!r"r3r<rArvrrrrs& (