&`i ddlZddlZddlZddlZddlZddlmZmZmZm Z m Z m Z m Z m Z mZmZmZmZddlZddlmZddlmZddlmZmZmZmZmZddlm Z m!Z!erddl"m#Z#GddeZ$ed Z%ed e$ Z&ed Z'ej(d Z)e de!GddZ*e!dGdde*ej+e e%e'fZ,e!Gdde,e-e-fZ.e!Gdde,ee-e/fee-e/ffZ0e!Gdde,e&e&fZ1e!Gdde,e&e&fZ2e!Gdde,e ee-e/fe/fZ3e!Gd d!e,e ee-e/fe/fZ4e!Gd"d#e,e&e&fZ5e!Gd$d%e,e ee efZ6e!Gd&d'e,eee efZ7e!Gd(d)e,Z8d*e9fd+Z:d,eege%fd*e9d-eege e%ffd.Z;d/ee%ge%fd-ee e%ge%ffd0Ze!dGd5d6e,e e-e/fZ?e!dGd7d8e,Z@e!dGd9d:e,ZAdS);N) TYPE_CHECKINGAnyCallable CollectionDictGenericListOptionalProtocolSetTypeVarUnionis_null)Block BlockAccessor BlockColumnBlockColumnAccessorKeyType) Deprecated PublicAPI)SchemacNeZdZdedefdZdedefdZdedefdZdedefdZdS)_SupportsRichComparisonotherreturncdSNselfrs f/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/aggregate.py__lt__z_SupportsRichComparison.__lt__' cdSrrr s r"__le__z_SupportsRichComparison.__le__*r$r%cdSrrr s r"__gt__z_SupportsRichComparison.__gt__-r$r%cdSrrr s r"__ge__z_SupportsRichComparison.__ge__0r$r%N) __name__ __module__ __qualname__rboolr#r'r)r+rr%r"rr&s C D     C D     C D     C D      r%rAccumulatorTypeSupportsRichComparisonType)bound AggOutputTypez^([^(]+)(?:\(.*\))?$z3AggregateFn is deprecated, please use AggregateFnV2)messageceZdZdZ ddeegefdeeegefdedeeeee fgefdeee gefde eege ff d Z d e d d dfd ZdS) AggregateFnae NOTE: THIS IS DEPRECATED, PLEASE USE :class:`AggregateFnV2` INSTEAD Defines how to perform a custom aggregation in Ray Data. `AggregateFn` instances are passed to a Dataset's ``.aggregate(...)`` method to specify the steps required to transform and combine rows sharing the same key. This enables implementing custom aggregators beyond the standard built-in options like Sum, Min, Max, Mean, etc. Args: init: Function that creates an initial aggregator for each group. Receives a key (the group key) and returns the initial accumulator state (commonly 0, an empty list, or an empty dictionary). merge: Function that merges two accumulators generated by different workers into one accumulator. name: An optional display name for the aggregator. Useful for debugging. accumulate_row: Function that processes an individual row. It receives the current accumulator and a row, then returns an updated accumulator. Cannot be used if `accumulate_block` is provided. accumulate_block: Function that processes an entire block of rows at once. It receives the current accumulator and a block of rows, then returns an updated accumulator. This allows for vectorized operations. Cannot be used if `accumulate_row` is provided. finalize: Function that finishes the aggregation by transforming the final accumulator state into the desired output. For example, if your accumulator is a list of items, you may want to compute a statistic from the list. If not provided, the final accumulator state is returned as-is. Example: .. testcode:: import ray from ray.data.aggregate import AggregateFn # A simple aggregator that counts how many rows there are per group count_agg = AggregateFn( init=lambda k: 0, accumulate_row=lambda counter, row: counter + 1, merge=lambda c1, c2: c1 + c2, name="custom_count" ) ds = ray.data.from_items([{"group": "A"}, {"group": "B"}, {"group": "A"}]) result = ds.groupby("group").aggregate(count_agg).take_all() # result: [{'group': 'A', 'custom_count': 2}, {'group': 'B', 'custom_count': 1}] Ninitmergenameaccumulate_rowaccumulate_blockfinalizec||td|dtdtdtffd }t|tst d|d}||_||_||_||_ ||_ dS)NzCExactly one of accumulate_row or accumulate_block must be provided.ablockrcztj|}|dD]}||}|SNF)public_row_format)r for_block iter_rows)r>r? block_accrr:s r"r;z.AggregateFn.__init__..accumulate_blocksK)3E:: ",,u,EE--A&q!,,AAr%z`name` must be provided.c|Srr)r>s r"z&AggregateFn.__init__..r%) ValueErrorr0r isinstancestr TypeErrorr9r7r8r;r<)r!r7r8r9r:r;r<s ` r"__init__zAggregateFn.__init__os  "'7'?  &+;+GU   # O E o       $$$ 8677 7  "{H   0  r%schemarrcdS)z=Raise an error if this cannot be applied to the given schema.Nr)r!rOs r" _validatezAggregateFn._validates r%)NNN)r,r-r.__doc__rrr0rLrrrr r3rNrQrr%r"r6r6=s--l PTIM$!$!y/12$!/:OKL$! $! ! d38n - > $!#OU#;_#LM$!8_$5}$DEF$!$!$!$!L  2 t      r%r6alpha) stabilityceZdZdZdedegefdeedeffd Z deefdZ defd Z e j d ed edefd Ze j d edefdZdedeefdZdedddfdZxZS) AggregateFnV2aProvides an interface to implement efficient aggregations to be applied to the dataset. `AggregateFnV2` instances are passed to a Dataset's ``.aggregate(...)`` method to perform distributed aggregations. To create a custom aggregation, you should subclass `AggregateFnV2` and implement the `aggregate_block` and `combine` methods. The `finalize` method can also be overridden if the final accumulated state needs further transformation. Aggregation follows these steps: 1. **Initialization**: For each group (if grouping) or for the entire dataset, an initial accumulator is created using `zero_factory`. 2. **Block Aggregation**: The `aggregate_block` method is applied to each block independently, producing a partial aggregation result for that block. 3. **Combination**: The `combine` method is used to merge these partial results (or an existing accumulated result with a new partial result) into a single, combined accumulator. 4. **Finalization**: Optionally, the `finalize` method transforms the final combined accumulator into the desired output format. Args: name: The name of the aggregation. This will be used as the column name in the output, e.g., "sum(my_col)". zero_factory: A callable that returns the initial "zero" value for the accumulator. For example, for a sum, this would be `lambda: 0`; for finding a minimum, `lambda: float("inf")`, for finding a maximum, `lambda: float("-inf")`. on: The name of the column to perform the aggregation on. If `None`, the aggregation is performed over the entire row (e.g., for `Count()`). ignore_nulls: Whether to ignore null values during aggregation. If `True`, nulls are skipped. If `False`, the presence of a null value might result in a null output, depending on the aggregation logic. r9 zero_factoryon ignore_nullsc |std|d||_||_t|}|r|d|_n||_t|j|}t|j | t|j }t||}t||| fd|dS)Nz1Non-empty string has to be provided as name (got )c|Srr)_r?_safe_aggregates r"rHz(AggregateFnV2.__init__..sooe.D.Dr%)r9r7r8r;r<)rJ_target_col_name _ignore_nulls_AGGREGATION_NAME_PATTERNmatchgroup _agg_name_null_safe_combinecombine_null_safe_aggregateaggregate_block_null_safe_finalizer<_null_safe_zero_factorysuperrN) r!r9rWrXrYrc _safe_combine_safe_finalize_safe_zero_factoryr_ __class__s @r"rNzAggregateFnV2.__init__s KDKKK !#)*//55  ""[[^^DNN!DN*4<FF .t/C\RR,T];;4\<PP #DDDD#      r%rc|jSr)r`r!s r"get_target_columnzAggregateFnV2.get_target_columns $$r%c|jS)zReturn the agg name (e.g., 'sum', 'mean', 'count'). Returns the aggregation type extracted from the name during initialization. For example, returns 'sum' for an aggregator named 'sum(col)'. )rerrs r" get_agg_namezAggregateFnV2.get_agg_names ~r%current_accumulatornewcdS)aCombines a new partial aggregation result with the current accumulator. This method defines how two intermediate aggregation states are merged. For example, if `aggregate_block` produces partial sums `s1` and `s2` from two different blocks, `combine(s1, s2)` should return `s1 + s2`. Args: current_accumulator: The current accumulated state (e.g., the result of previous `combine` calls or an initial value from `zero_factory`). new: A new partially aggregated value, typically the output of `aggregate_block` from a new block of data, or another accumulator from a parallel task. Returns: The updated accumulator after combining it with the new value. Nrr!rvrws r"rgzAggregateFnV2.combines ( r%r?cdS)aAggregates data within a single block. This method processes all rows in a given `Block` and returns a partial aggregation result for that block. For instance, if implementing a sum, this method would sum all relevant values within the block. Args: block: A `Block` of data to be aggregated. Returns: A partial aggregation result for the input block. The type of this result (`AggType`) should be consistent with the `current_accumulator` and `new` arguments of the `combine` method, and the `accumulator` argument of the `finalize` method. Nrr!r?s r"rizAggregateFnV2.aggregate_blocks " r% accumulatorc|S)aTransforms the final accumulated state into the desired output. This method is called once per group after all blocks have been processed and all partial results have been combined. It provides an opportunity to perform a final transformation on the accumulated data. For many aggregations (e.g., Sum, Count, Min, Max), the accumulated state is already the final result, so this method can simply return the accumulator as is (which is the default behavior). For other aggregations, like Mean, this method is crucial. A Mean aggregation might accumulate `[sum, count]`. The `finalize` method would then compute `sum / count` to get the final mean. Args: accumulator: The final accumulated state for a group, after all `aggregate_block` and `combine` operations. Returns: The final result of the aggregation for the group. rr!r|s r"r<zAggregateFnV2.finalizes ,r%rOrNcj|jr+ddlm}||j|dSdS)Nr)SortKey)r`2ray.data._internal.planner.exchange.sort_task_specrvalidate_schema)r!rOrs r"rQzAggregateFnV2._validate2sS   C R R R R R R GD) * * : :6 B B B B B C Cr%)r,r-r.rRrLrr0r r/rNrsruabcabstractmethodrgrrir3r<rQ __classcell__rps@r"rVrVs}""H$ $ r?23$ SM $  $ $ $ $ $ $ L%8C=%%%%c  #2 9H     *  U     $O8O0C 2CtCCCCCCCCr%rVcteZdZdZ ddeededeeffd Zded e fd Z d e d e d e fd Z xZ S)CountaDefines count aggregation. Example: .. testcode:: import ray from ray.data.aggregate import Count ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Counting all rows: result = ds.aggregate(Count()) # result: {'count()': 100} # Counting all rows per group: result = ds.groupby("group_key").aggregate(Count(on="id")).take_all() # result: [{'group_key': 0, 'count(id)': 34}, # {'group_key': 1, 'count(id)': 33}, # {'group_key': 2, 'count(id)': 33}] Args: on: Optional name of the column to count values on. If None, counts rows. ignore_nulls: Whether to ignore null values when counting. Only applies if `on` is specified. Default is `False` which means `Count()` on a column will count nulls by default. To match pandas default behavior of not counting nulls, set `ignore_nulls=True`. alias_name: Optional name for the resulting column. NFrXrY alias_namecht|r|nd|pdd||ddS)Nzcount(r[cdSNrrrr%r"rHz Count.__init__..hrIr%rXrYrW)rlrNr!rXrYrrps r"rNzCount.__init__^sW $ >JJ*>28*>*>*>%"      r%r?rctj|}|j|S||j|jS)NrY)rrCr`num_rowscountra)r!r?block_accessors r"rizCount.aggregate_blockksU&077  (!**,, ,##  !0B$   r%rvrwc ||zSrrrys r"rgz Count.combinevs "S((r%)NFN) r,r-r.rRr rLr/rNrintrirgrrs@r"rr9s!!J!"$(   SM    SM         U  s     )3)S)S))))))))r%rceZdZdZ ddeededeeffd Zded e e e ffd Z d e e e fd e e e fd e e e ffd Z xZS)Suma7Defines sum aggregation. Example: .. testcode:: import ray from ray.data.aggregate import Sum ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Summing all rows per group: result = ds.aggregate(Sum(on="id")) # result: {'sum(id)': 4950} Args: on: The name of the numerical column to sum. Must be provided. ignore_nulls: Whether to ignore null values during summation. If `True` (default), nulls are skipped. If `False`, the sum will be null if any value in the group is null. alias_name: Optional name for the resulting column. NTrXrYrc~t|r|ndt|d||ddS)Nzsum(r[cdSrrrr%r"rHzSum.__init__..rIr%rrlrNrLrs r"rNz Sum.__init__sY $ ;JJ*;R*;*;*;%"      r%r?rcftj||j|jSr)rrCsumr`rar{s r"rizSum.aggregate_block0&u--11  !4#5   r%rvrwc ||zSrrrys r"rgz Sum.combines#S((r%NTN)r,r-r.rRr rLr/rNrrrfloatrirgrrs@r"rrzs8!!$(   SM    SM        U uS%Z/@    )#(e#4);@e;L) sEz ))))))))r%rc eZdZdZddddfdeededeedegefffd Z d e d efd Z d eded efdZ xZ S)Mina3Defines min aggregation. Example: .. testcode:: import ray from ray.data.aggregate import Min ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Finding the minimum value per group: result = ds.groupby("group_key").aggregate(Min(on="id")).take_all() # result: [{'group_key': 0, 'min(id)': 0}, # {'group_key': 1, 'min(id)': 1}, # {'group_key': 2, 'min(id)': 2}] Args: on: The name of the column to find the minimum value from. Must be provided. ignore_nulls: Whether to ignore null values. If `True` (default), nulls are skipped. If `False`, the minimum will be null if any value in the group is null (for most data types, or follow type-specific comparison rules with nulls). alias_name: Optional name for the resulting column. zero_factory: A callable that returns the initial "zero" value for the accumulator. For example, for a float column, this would be `lambda: float("+inf")`. Default is `lambda: float("+inf")`. NTc tdS)Nz+infrrr%r"rHz Min. vr%rXrYrrWc|t|r|ndt|d|||dS)Nzmin(r[rrr!rXrYrrWrps r"rNz Min.__init__W $ ;JJ*;R*;*;*;%%      r%r?rcftj||j|jSr)rrCminr`rar{s r"rizMin.aggregate_blockrr%rvrwc"t||Sr)rrys r"rgz Min.combine &,,,r%r,r-r.rRr rLr/rr1rNrrirgrrs@r"rrD!!$(AVAV   SM    SM  r#==>        U /I    -7-(- $ --------r%rc eZdZdZddddfdeededeedegefffd Z d e d efd Z d eded efdZ xZ S)Maxa9Defines max aggregation. Example: .. testcode:: import ray from ray.data.aggregate import Max ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Finding the maximum value per group: result = ds.groupby("group_key").aggregate(Max(on="id")).take_all() # result: [{'group_key': 0, 'max(id)': ...}, # {'group_key': 1, 'max(id)': ...}, # {'group_key': 2, 'max(id)': ...}] Args: on: The name of the column to find the maximum value from. Must be provided. ignore_nulls: Whether to ignore null values. If `True` (default), nulls are skipped. If `False`, the maximum will be null if any value in the group is null (for most data types, or follow type-specific comparison rules with nulls). alias_name: Optional name for the resulting column. zero_factory: A callable that returns the initial "zero" value for the accumulator. For example, for a float column, this would be `lambda: float("-inf")`. Default is `lambda: float("-inf")`. NTc tdS)Nz-infrrr%r"rHz Max.rr%rXrYrrWc|t|r|ndt|d|||dS)Nzmax(r[rrrs r"rNz Max.__init__ rr%r?rcftj||j|jSr)rrCmaxr`rar{s r"rizMax.aggregate_blockrr%rvrwc"t||Srrrys r"rgz Max.combine rr%rrs@r"rrrr%rc (eZdZdZ ddeededeeffd Zded ee e e e ffd Z d e e e e fd e e e e fd e e e e ffd Zde e e e fd ee fdZxZS)MeanaDefines mean (average) aggregation. Example: .. testcode:: import ray from ray.data.aggregate import Mean ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Calculating the mean value per group: result = ds.groupby("group_key").aggregate(Mean(on="id")).take_all() # result: [{'group_key': 0, 'mean(id)': ...}, # {'group_key': 1, 'mean(id)': ...}, # {'group_key': 2, 'mean(id)': ...}] Args: on: The name of the numerical column to calculate the mean on. Must be provided. ignore_nulls: Whether to ignore null values. If `True` (default), nulls are skipped. If `False`, the mean will be null if any value in the group is null. alias_name: Optional name for the resulting column. NTrXrYrc~t|r|ndt|d||ddS)Nzmean(r[c$tddgSrlistrr%r"rHzMean.__init__..Ssq!fr%rrrs r"rNz Mean.__init__Fs\ $ Q  6M1~ A..r%r)r,r-r.rRr rLr/rNrr rrrrirgr<rrs@r"rr(sB<!!$(   SM  SM       UxU3:=N8O/P$R#'c5j(9#:RAEeCQVJFWAXR eCJ RRRR /DsEz):$;/////////r%rc eZdZdZ ddeedededeeffd Zd e d e e ee ffd Z d e e de e d e e fdZde e d ee fdZxZS)StdahDefines standard deviation aggregation. Uses Welford's online algorithm for numerical stability. This method computes the standard deviation in a single pass. Results may differ slightly from libraries like NumPy or Pandas that use a two-pass algorithm but are generally more accurate. See: https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm Example: .. testcode:: import ray from ray.data.aggregate import Std ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Calculating the standard deviation per group: result = ds.groupby("group_key").aggregate(Std(on="id")).take_all() # result: [{'group_key': 0, 'std(id)': ...}, # {'group_key': 1, 'std(id)': ...}, # {'group_key': 2, 'std(id)': ...}] Args: on: The name of the column to calculate standard deviation on. ddof: Delta Degrees of Freedom. The divisor used in calculations is `N - ddof`, where `N` is the number of elements. Default is 1. ignore_nulls: Whether to ignore null values. Default is True. alias_name: Optional name for the resulting column. Nr\TrXddofrYrct|r|ndt|d||d||_dS)Nzstd(r[c$tgdS)N)rrrrrr%r"rHzStd.__init__..siiir%r)rlrNrL_ddof)r!rXrrYrrps r"rNz Std.__init__s_ $ ;JJ*;R*;*;*;% 10   r%r?rc8tj|}||j|j}|dks|dS||j|j}t |r|S||z }||j|j|}|||gSNrr)rrCrr`rarrsum_of_squared_diffs_from_mean)r!r?rErrmeanM2s r"rizStd.aggregate_blocks!+E22  5DDVWW A::4}}T2D4FGG 4== Ke|  5 5  !4#5t  D%  r%rvrwc||\}}}|\}}}||z } ||z} ||z||zz| z } ||z| dz|z|z| z z} | | | gS)Nr) r!rvrwM2_amean_acount_aM2_bmean_bcount_bdeltarrrs r"rgz Std.combinesx !4fg #fg'!  6G#33u< D[E1H/'9EA AD%  r%r|c|\}}}||jz dkr tjStj|||jz z Sr)rrrmathsqrt)r!r|rrrs r"r<z Std.finalizesF&D% 4:  " "6Myutz12333r%)Nr\TN)r,r-r.rRr rLrr/rNrr rrrirgr<rrs@r"rrws!!J!!$(  SM  SM *!U!tE#u*4E/F!!!!"!#';!59%[! e!!!!$ 4DK 4HUO 4 4 4 4 4 4 4 4r%rc eZdZdZddddfdeededeedegefffd Z d e d eefd Z d eded efdZ xZ S)AbsMaxa_Defines absolute max aggregation. Example: .. testcode:: import ray from ray.data.aggregate import AbsMax ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Calculating the absolute maximum value per group: result = ds.groupby("group_key").aggregate(AbsMax(on="id")).take_all() # result: [{'group_key': 0, 'abs_max(id)': ...}, # {'group_key': 1, 'abs_max(id)': ...}, # {'group_key': 2, 'abs_max(id)': ...}] Args: on: The name of the column to calculate absolute maximum on. Must be provided. ignore_nulls: Whether to ignore null values. Default is True. alias_name: Optional name for the resulting column. zero_factory: A callable that returns the initial "zero" value for the accumulator. For example, for a float column, this would be `lambda: 0`. Default is `lambda: 0`. NTcdSrrrr%r"rHzAbsMax.sr%rXrYrrWc|t|tstd|dt|r|ndt|d|||dS)Nz/Column to aggregate on has to be provided (got r[zabs_max(r)rKrLrJrlrNrs r"rNzAbsMax.__init__s :ZC00:TrTTTUU U $ ?JJ*?SWW*?*?*?%%      r%r?rc>tj|}||j|j}||j|j}t |st |rdStt|t|Sr)rrCrr`rarrabs)r!r?rmax_min_s r"rizAbsMax.aggregate_blocks&077!!$"79KLL!!$"79KLL 4== GDMM 43t99c$ii(((r%rvrwc"t||Srrrys r"rgzAbsMax.combinerr%rrs@r"rrs>!!$(AJ   SM  SM  r#==>       " )U )x8R/S ) ) ) )-7-(- $ --------r%rc eZdZdZ ddeedededeeffd Zd e e d e e d e e fd Z de d e e fdZ de e d ee fdZxZS)Quantilea Defines Quantile aggregation. Example: .. testcode:: import ray from ray.data.aggregate import Quantile ds = ray.data.range(100) # Schema: {'id': int64} ds = ds.add_column("group_key", lambda x: x % 3) # Schema: {'id': int64, 'group_key': int64} # Calculating the 50th percentile (median) per group: result = ds.groupby("group_key").aggregate(Quantile(q=0.5, on="id")).take_all() # result: [{'group_key': 0, 'quantile(id)': ...}, # {'group_key': 1, 'quantile(id)': ...}, # {'group_key': 2, 'quantile(id)': ...}] Args: on: The name of the column to calculate the quantile on. Must be provided. q: The quantile to compute, which must be between 0 and 1 inclusive. For example, q=0.5 computes the median. ignore_nulls: Whether to ignore null values. Default is True. alias_name: Optional name for the resulting column. N?TrXqrYrc||_t|r|ndt|d||tdS)Nz quantile(r[r)_qrlrNrLr)r!rXrrYrrps r"rNzQuantile.__init__Cs^ $ @JJ*@c"gg*@*@*@%      r%rvrwrc$t|tr,t|tr|||St|tr4t|ts||dkr|||St|tr4t|ts||dkr|||Sg}||dkr||||dkr|||S)Nr)rKr extendappend)r!rvrwlss r"rgzQuantile.combineSs" )4 0 0 'ZT5J5J '  & &s + + +& & )4 0 0 '*S$:O:O '3"99#**3///& & c4  *5H$*O*O ".3F"3L3L .///J   */Bb/H/H II) * * * ?sbyy IIcNNN r%r?ctj|}g}|dD]/}|||j0|SrA)rrCrDrgetr`)r!r?rErrows r"rizQuantile.aggregate_blockls`!+E22 &&&?? 6 6C IIcggd344 5 5 5 5 r%r|c&|jr d|D}n'd|D}t|dkr|dS|sdSd}t|}t|dz |jz}t j|}t j|}||kr||t|S||t|||z z}||t|||z z} t|| zdS)Nc0g|]}t||Srr.0vs r" z%Quantile.finalize..ws#DDDD1DDDr%c0g|]}t||Srrrs r"rz%Quantile.finalize..ys#:::1wqzz:Q:::r%rc|Srrxs r"rHz#Quantile.finalize..sr%r\) ralensortedrrfloorceilrround) r!r|nullskey input_valueskfcd0d1s r"r<zQuantile.finalizeus%   DDkDDDKK:: :::E5zzA~~Qx 4kk**    "dg - JqMM IaLL 663|CFF+,, ,Sc!ff% & &!a% 0 Sc!ff% & &!a% 0R"Wa   r%)NrTN)r,r-r.rRr rLrr/rNr rrgrrir<rrs@r"rr%s <!!$(   SM     SM       4949c2UtCy!DI!(3-!!!!!!!!r%rc *eZdZdZGddeejZ ddeede deed e e edfffd Z d e e d e e d e e fdZded efdZded ee fdZedZeded e fdZxZS)UniqueabDefines unique aggregation. Example: .. testcode:: import ray from ray.data.aggregate import Unique ds = ray.data.range(100) ds = ds.add_column("group_key", lambda x: x % 3) # Calculating the unique values per group: result = ds.groupby("group_key").aggregate(Unique(on="id")).take_all() # result: [{'group_key': 0, 'unique(id)': ...}, # {'group_key': 1, 'unique(id)': ...}, # {'group_key': 2, 'unique(id)': ...}] Args: on: The name of the column from which to collect unique values. ignore_nulls: Whether to ignore null values when collecting unique items. Default is True (nulls are excluded). alias_name: Optional name for the resulting column. encode_lists: If `True`, encode list elements. If `False`, encode whole lists (i.e., the entire list is considered as a single object). `False` by default. Note that this is a top-level flatten (not a recursive flatten) operation. ceZdZdZdZdZdS)Unique.ListEncodingModea_Controls how to encode individual elements inside the list column: - NONE: no encoding applied, elements (lists) are stored as is and unique ones are returned. - FLATTEN: column of element lists is flattened into a single list. - HASH: each list element is hashed, a list of unique hashes is returned. FLATTENHASHN)r,r-r.rRrrrr%r"ListEncodingModer s$  r%rNFrXrYr encode_listsc8t|r|ndt|d||tt |t jr ||_dSt |tr|rt jj |_dSd|_dS)Nzunique(r[r) rlrNrLsetrKr r_list_encoding_moder/r)r!rXrYrrrps r"rNzUnique.__init__s $ >JJ*>CGG*>*>*>%     lF$; < < ,'3D $ $ $  d + + ,  ,'-'>'FD $ $ $'+D $ $ $r%rvrwrcX||||zSr)_to_setrys r"rgzUnique.combines&||/004<<3D3DDDr%r?c.||j}tj|}|r|j|jt jjkr'tj|}nX|jt jj kr'tj| }ntd|j|j r&tj| }|S)Nz"list encoding mode not supported: )r`r for_columnis_composed_of_listsrr rrflattenrhashrJradropnaunique)r!r?columncolumn_accessors r"_compute_uniquezUnique._compute_uniquest,--8@@  0 0 2 2 (4'6+B+JJJ"5"@#++--##)V-D-III"5"@AUAUAWAW"X"X S9QSS   W1<_=S=S=U=UVVO%%'''r%cx||}tj|Sr)r rr to_pylist)r!r?rs r"rizUnique.aggregate_blocks2%%e,,"-f55??AAAr%cBt|trt|St|trYt |dkr,t|dtrt d|}t|S|hS)Nrc(|dnt|Sr)tuple)rs r"rHz Unique._to_set..s!)$$qr%)rKrr _normalize_nansrrmaprs r"rzUnique._to_sets a   ))!,, , 4  1vvzzj1t44zAA1EE))!,, ,3Jr%rcd|DS)Nc|h|]9}t|trtj|s|n tj:Sr)rKrrisnanrrs r" z)Unique._normalize_nans..s;WWWPQ*Q..K28A;;KRVWWWr%rrs r"r&zUnique._normalize_nanss XWUVWWWWr%)NFNN)r,r-r.rRrLenumEnumrr r/rrNr rrgrrr r ri staticmethodrrr&rrs@r"r r s:     3   !"$(<@ ,, SM,,SM , D"2D89 ,,,,,,*E3s8E#c(Es3xEEEE(U({((((0BUBtCyBBBB  \ X:X#XXX\XXXXXr%r ceZdZdZ d dedeeffd Zdedeee ffdZ d eee fd eee fdeee ffd Z xZ S) ValueCounteraCounts the number of times each value appears in a column. This aggregation computes value counts for a specified column, similar to pandas' `value_counts()` method. It returns a dictionary with two lists: "values" containing the unique values found in the column, and "counts" containing the corresponding count for each value. Example: .. testcode:: import ray from ray.data.aggregate import ValueCounter # Create a dataset with repeated values ds = ray.data.from_items([ {"category": "A"}, {"category": "B"}, {"category": "A"}, {"category": "C"}, {"category": "A"}, {"category": "B"} ]) # Count occurrences of each category result = ds.aggregate(ValueCounter(on="category")) # result: {'value_counter(category)': {'values': ['A', 'B', 'C'], 'counts': [3, 2, 1]}} # Using with groupby ds = ray.data.from_items([ {"group": "X", "category": "A"}, {"group": "X", "category": "B"}, {"group": "Y", "category": "A"}, {"group": "Y", "category": "A"} ]) result = ds.groupby("group").aggregate(ValueCounter(on="category")).take_all() # result: [{'group': 'X', 'value_counter(category)': {'values': ['A', 'B'], 'counts': [1, 1]}}, # {'group': 'Y', 'value_counter(category)': {'values': ['A'], 'counts': [2]}}] Args: on: The name of the column to count values in. Must be provided. alias_name: Optional name for the resulting column. If not provided, defaults to "value_counter({column_name})". NrXrc~t|r|ndt|d|dddS)Nzvalue_counter(r[Tc ggdS)N)valuescountsrrr%r"rHz'ValueCounter.__init__..=sB"!=!=r%rrr!rXrrps r"rNzValueCounter.__init__4sZ $ EJJ*E3r77*E*E*E==      r%r?rchtj||j}|Sr)rrr` value_counts)r!r? col_accessors r"rizValueCounter.aggregate_block@s,*5eDz(ValueCounter.combine..Os===41a!Q===r%) enumerateziprr) r!rvr9r3r4value_to_indexv_newc_newidxs r"rgzValueCounter.combineEs %X.$X.>=9V+<+<=== 9?8;TUU % %LE5&&$U+s u$ (+F u% e$$$ e$$$$""r%r) r,r-r.rRrLr rNrrr rirgrrs@r"r0r0 s%%T%)     SM       +U+tCI++++ #!#t)_#c4i# c4i ########r%r0rYc|rd}nfd}|S)aNOTE: PLEASE READ CAREFULLY BEFORE CHANGING Null-safe zero factory is crucial for implementing proper aggregation protocol (monoid) w/o the need for additional containers. Main hurdle for implementing proper aggregation semantic is to be able to encode semantic of an "empty accumulator" and be able to tell it from the case when accumulator is actually holding null value: - Empty container can be overridden with any value - Container holding null can't be overridden if ignore_nulls=False However, it's possible for us to exploit asymmetry in cases of ignore_nulls being True or False: - Case of ignore_nulls=False entails that if there's any "null" in the sequence, aggregation is undefined and correspondingly expected to return null - Case of ignore_nulls=True in turn, entails that if aggregation returns "null" if and only if the sequence does NOT have any non-null value Therefore, we apply this difference in semantic to zero-factory to make sure that our aggregation protocol is adherent to that definition: - If ignore_nulls=True, zero-factory returns null, therefore encoding empty container - If ignore_nulls=False, couldn't return null as aggregation will incorrectly prioritize it, and instead it returns true zero value for the aggregation (ie 0 for count/sum, -inf for max, etc). cdSrr)r^s r"roz3_null_safe_zero_factory.._safe_zero_factorys4r%cSrr)r^rWs r"roz3_null_safe_zero_factory.._safe_zero_factorys<>> !r%r)rWrYros` r"rkrk]sC@"      " " " " " r% aggregatercJdtdttffd }|S)Nr?rcD|}t|rrdS|Srr)r?resultrGrYs r"r_z-_null_safe_aggregate.._safe_aggregates35!! 6?? | 4 r%)rr r0)rGrYr_s`` r"rhrhsBu/)B r%r<cFdttdtffd }|S)Naccrc<t|r|n |Srr)rLr<s r"rnz+_null_safe_finalize.._safe_finalizes"cll5ss 5r%r r0)r<rns` r"rjrjs<6H_56/666666 r%rgc|r=dttdttdttffd }n._safe_combine=s|| )  ) wsC(((r%c`t|r|St|r|S||SrrrRs r"rmz)_null_safe_combine.._safe_combinerSr%rN)rgrYrms` r"rfrfs2) )/* )19/1J ) o & ) ) ) ) ) ) ) )/* )19/1J ) o & ) ) ) ) ) ) r%ceZdZdZ ddedeeffd Zdedee fdZ d ee d ee dee fd Z d ee dee fd Z xZS)MissingValuePercentageaCalculates the percentage of null values in a column. This aggregation computes the percentage of null (missing) values in a dataset column. It treats both None values and NaN values as null. The result is a percentage value between 0.0 and 100.0, where 0.0 means no missing values and 100.0 means all values are missing. Example: .. testcode:: import ray from ray.data.aggregate import MissingValuePercentage # Create a dataset with some missing values ds = ray.data.from_items([ {"value": 1}, {"value": None}, {"value": 3}, {"value": None}, {"value": 5} ]) # Calculate missing value percentage result = ds.aggregate(MissingValuePercentage(on="value")) # result: 40.0 (2 out of 5 values are missing) # Using with groupby ds = ray.data.from_items([ {"group": "A", "value": 1}, {"group": "A", "value": None}, {"group": "B", "value": 3}, {"group": "B", "value": None} ]) result = ds.groupby("group").aggregate(MissingValuePercentage(on="value")).take_all() # result: [{'group': 'A', 'missing_pct(value)': 50.0}, # {'group': 'B', 'missing_pct(value)': 50.0}] Args: on: The name of the column to calculate missing value percentage on. alias_name: Optional name for the resulting column. If not provided, defaults to "missing_pct({column_name})". NrXrc~t|r|ndt|d|dddS)Nz missing_pct(r[Fc ddgSrrrr%r"rHz1MissingValuePercentage.__init__..  !Qr%rrr5s r"rNzMissingValuePercentage.__init__sY $ CJJ*CR*C*C*C'      r%r?rc tj||j}|d}t jt j|d}||gS)NFrT) nan_is_null) rrr`rpcrr_as_arrow_compatibleas_py)r!r?r total_count null_counts r"riz&MissingValuePercentage.aggregate_blocksx-8t?T9UVV%+++?? V J;;==4 P P P  %''  K((r%rvrwct|t|cxkrdksnJ|d|dz|d|dzgS)Nrrr\)rrys r"rgzMissingValuePercentage.combinesb&''3s888888q888888  "SV +  "SV +  r%r|cF|ddkrdS|d|dz dzSNr\rgY@rr~s r"r<zMissingValuePercentage.finalize#s. q>Q  4AQ/588r%r)r,r-r.rRrLr rNrr rrirgrr<rrs@r"rVrVs%%T%)     SM        )U )tCy ) ) ) ) 49 49 c    9DI9(5/99999999r%rVceZdZdZ ddededeeffd Zded e e fd Z d e e d e e d e e fd Z de e d ee fdZxZS)ZeroPercentageaCalculates the percentage of zero values in a numeric column. This aggregation computes the percentage of zero values in a numeric dataset column. It can optionally ignore null values when calculating the percentage. The result is a percentage value between 0.0 and 100.0, where 0.0 means no zero values and 100.0 means all non-null values are zero. Example: .. testcode:: import ray from ray.data.aggregate import ZeroPercentage # Create a dataset with some zero values ds = ray.data.from_items([ {"value": 0}, {"value": 1}, {"value": 0}, {"value": 3}, {"value": 0} ]) # Calculate zero value percentage result = ds.aggregate(ZeroPercentage(on="value")) # result: 60.0 (3 out of 5 values are zero) # With null values and ignore_nulls=True (default) ds = ray.data.from_items([ {"value": 0}, {"value": None}, {"value": 0}, {"value": 3}, {"value": 0} ]) result = ds.aggregate(ZeroPercentage(on="value", ignore_nulls=True)) # result: 75.0 (3 out of 4 non-null values are zero) # Using with groupby ds = ray.data.from_items([ {"group": "A", "value": 0}, {"group": "A", "value": 1}, {"group": "B", "value": 0}, {"group": "B", "value": 0} ]) result = ds.groupby("group").aggregate(ZeroPercentage(on="value")).take_all() # result: [{'group': 'A', 'zero_pct(value)': 50.0}, # {'group': 'B', 'zero_pct(value)': 100.0}] Args: on: The name of the column to calculate zero value percentage on. Must be a numeric column. ignore_nulls: Whether to ignore null values when calculating the percentage. If True (default), null values are excluded from both numerator and denominator. If False, null values are included in the denominator but not the numerator. alias_name: Optional name for the resulting column. If not provided, defaults to "zero_pct({column_name})". TNrXrYrc~t|r|ndt|d||ddS)Nz zero_pct(r[c ddgSrrrr%r"rHz)ZeroPercentage.__init__..krYr%rrrs r"rNzZeroPercentage.__init__`sY $ @JJ*@c"gg*@*@*@%'      r%r?rc4tj||j}||j}|dkrddgS|}t j|d}t j| pd}||gSr) rrr`rrar]r\equalrr^)r!r?rrarrow_compatible zero_mask zero_counts r"rizZeroPercentage.aggregate_blockns-8t?T9UVV%%43E%FF A::q6M*??AAH-q11 VI&&,,..3! E""r%rvrwcF|d|dz|d|dzgSrrrys r"rgzZeroPercentage.combines.  "SV +  "SV +  r%r|cF|ddkrdS|d|dz dzSrcrr~s r"r<zZeroPercentage.finalizes. q>Q  4AQ/588r%)TN)r,r-r.rRrLr/r rNrr rrirgrr<rrs@r"rere*s22n"$(       SM       #U#tCy####$ 49 49 c    9DI9(5/99999999r%rec eZdZdZ ddedeededeeffd Z defd Z d e d e fd Z d e de d e fdZde d eefdZxZS)ApproximateQuantilecZ ddlm}n"#t$r}td|d}~wwxYw|S)Nr)kll_floats_sketchzdApproximateQuantile requires the `datasketches` package. Install it with `pip install datasketches`.) datasketchesrr ImportError)r!rrexcs r"_require_datasketchesz)ApproximateQuantile._require_datasketchess_  6 6 6 6 6 6 6   >   !  (#( NrX quantilesquantile_precisionrc_|__t |r|ndt |d|dfddS)u Computes the approximate quantiles of a column by using a datasketches kll_floats_sketch. https://datasketches.apache.org/docs/KLL/KLLSketch.html The accuracy of the KLL quantile sketch is a function of the configured quantile precision, which also affects the overall size of the sketch. The KLL Sketch has absolute error. For example, a specified rank accuracy of 1% at the median (rank = 0.50) means that the true quantile (if you could extract it from the set) should be between getQuantile(0.49) and getQuantile(0.51). This same 1% error applied at a rank of 0.95 means that the true quantile should be between getQuantile(0.94) and getQuantile(0.96). In other words, the error is a fixed +/- epsilon for the entire range of ranks. Typical single-sided rank error by quantile_precision (use for getQuantile/getRank): - quantile_precision=100 → ~2.61% - quantile_precision=200 → ~1.33% - quantile_precision=400 → ~0.68% - quantile_precision=800 → ~0.35% See https://datasketches.apache.org/docs/KLL/KLLAccuracyAndSize.html for details on accuracy and size. Null values in the target column are ignored when constructing the sketch. Example: .. testcode:: import ray from ray.data.aggregate import ApproximateQuantile # Create a dataset with some values ds = ray.data.from_items( [{"value": 20.0}, {"value": 40.0}, {"value": 60.0}, {"value": 80.0}, {"value": 100.0}] ) result = ds.aggregate(ApproximateQuantile(on="value", quantiles=[0.1, 0.5, 0.9])) # Result: {'approx_quantile(value)': [20.0, 60.0, 100.0]} Args: on: The name of the column to calculate the quantile on. Must be a numeric column. quantiles: The list of quantiles to compute. Must be between 0 and 1 inclusive. For example, quantiles=[0.5] computes the median. Null entries in the source column are skipped. quantile_precision: Controls the accuracy and memory footprint of the sketch (K in KLL); higher values yield lower error but use more memory. Defaults to 800. See https://datasketches.apache.org/docs/KLL/KLLAccuracyAndSize.html for details on accuracy and size. alias_name: Optional name for the resulting column. If not provided, defaults to "approx_quantile({column_name})". zapprox_quantile(r[TcRSrzero serialize)rzr!sr"rHz.ApproximateQuantile.__init__..s +=!>!>!H!H!J!Jr%rN)rv _sketch_cls _quantiles_quantile_precisionrlrNrL)r!rXryrzrrps` ` r"rNzApproximateQuantile.__init__sh 5577##5  $ GJJ*GSWW*G*G*GJJJJJ      r%c.||S)N)r)r)r!rzs r"r~zApproximateQuantile.zeros"4555r%r?rctj|}|}||}||j}|D]J}|4|t|K| Sr) rrCto_arrowrrsr~rr^updaterr)r!r?rEtablersketchvalues r"riz#ApproximateQuantile.aggregate_blocks!+E22 ""$$d4466774344 4 4E{{}}( eEKKMM22333!!!r%rvrwc||j}||j|||j||Sr)r~rr8r deserializerr!rvrwcombineds r"rgzApproximateQuantile.combineso99T566t'334GHHIIIt'33C88999!!###r%r|cf|j||jSr)rr get_quantilesrr~s r"r<zApproximateQuantile.finalizes)++K88FFtWWWr%)rxN)r,r-r.rvrLr rrr rNr~rbytesrirgr<rrs@r"rprps!!!#&$( < < < ;<  < SM < < < < < < |6s6666 "U "u " " " "$5$u$$$$$ XEXd5kXXXXXXXXr%rpc eZdZdZ ddedededeed ef fd Zdefd Z d e d e fdZ de de d e fdZ de d eeeeffdZxZS)ApproximateTopKcZ ddlm}n"#t$r}td|d}~wwxYw|S)Nr)frequent_strings_sketchz`ApproximateTopK requires the `datasketches` package. Install it with `pip install datasketches`.)rsrrt)r!rrus r"rvz%ApproximateTopK._require_datasketchess_  < < < < < < <   >   '&rwNFrXr log_capacityrrc|___|_t |r|ndt|d|dfddS)u_ Computes the approximate top k items in a column by using a datasketches frequent_strings_sketch. https://datasketches.apache.org/docs/Frequency/FrequentItemsOverview.html Guarantees: - Any item with true frequency > N / (2^log_capacity) is guaranteed to appear in the results - Reported counts may have an error of at most ± N / (2^log_capacity). If log_capacity is too small for your data: - Low-frequency items may be evicted from the sketch, potentially causing the top-k results to miss items that should appear in the output. - The error bounds increase, reducing the accuracy of the reported counts. Example: .. testcode:: import ray from ray.data.aggregate import ApproximateTopK ds = ray.data.from_items([ {"word": "apple"}, {"word": "banana"}, {"word": "apple"}, {"word": "cherry"}, {"word": "apple"} ]) result = ds.aggregate(ApproximateTopK(on="word", k=2)) # Result: {'approx_topk(word)': [{'word': 'apple', 'count': 3}, {'word': 'banana', 'count': 1}]} Args: on: The name of the column to aggregate. k: The number of top items to return. log_capacity: Base 2 logarithm of the maximum size of the internal hash map. Higher values increase accuracy but use more memory. Defaults to 15. alias_name: The name of the aggregate. Defaults to None. encode_lists: If `True`, encode list elements. If `False`, encode whole lists (i.e., the entire list is considered as a single object). `False` by default. Note that this is a top-level flatten (not a recursive flatten) operation. z approx_topk(r[TcRSrr})rr!sr"rHz*ApproximateTopK.__init__..4s._sV   V\%--Q"8"8997CQLL Q   r%)rsrrsrrget_frequent_itemsNO_FALSE_NEGATIVESr)r!r|rfrequent_itemsrs @r"r<zApproximateTopK.finalizeVs::::::''))6BB    6I J J     &xx0    r%)rNF)r,r-r.rvrLrr r/rNr~rrrirgr rrr<rrs@r"rrs+'''$(" ; ; ;  ;  ; SM ;  ; ; ; ; ; ; zDDDDD"U"u"""",$5$u$$$$$  E  d4S>.B         r%r)Brr,rrretypingrrrrrrr r r r r rnumpyrpyarrow.computecomputer\ray.data._internal.utilrray.data.blockrrrrrray.util.annotationsrrray.data.datasetrrr0r1r3compilerbr6ABCrVrrrrrrrrrrr r0r/rkrhrjrfrVrerprrr%r"rs                             ++++++76666666(''''''      h    '+,,$W (?(( &BJ'>?? IJJJ X X X X X X X  KJX v W[C[C[C[C[CK'/=2P*Q[C[C[C| =)=)=)=)=)M#s( #=)=) =)@ 0)0)0)0)0)-c5j)5e+<< =0)0) 0)f 9-9-9-9-9--24NN O9-9- 9-x 9-9-9-9-9--24NN O9-9- 9-x K/K/K/K/K/=eCJ/0%7 8K/K/ K/\ g4g4g4g4g4-U3:./6 7g4g4 g4T ?-?-?-?-?-]57QQ R?-?- ?-D k!k!k!k!k!}T#YS 12k!k! k!\ sXsXsXsXsX]3s8T#Y. /sXsX sXl N#N#N#N#N#=N#N# N#b*****Z 01  ugx001     (/9: x()?:;1 8/I J11 o 9:H_