&`ilddlmZddlmZddlmZmZmZm Z mZm Z m Z m Z m Z ddlmZddlmZddlmZddlmZmZmZmZmZmZmZddlmZmZmZm Z m!Z!dd l"m#Z#dd l$m%Z%m&Z&dd l'm(Z(m)Z)m*Z*dd l+m,Z,d Z-dZ.GddZ/de ee dge fee dgee ffdede e0de e0dededee fdZ1e/Z2dS))Iterator)partial) AnyCallableDictIterablerListOptionalTupleUnion)ComputeStrategy) LogicalPlan) Aggregate) AggregateFnCountMaxMeanMinStdSum)Block BlockAccessor CallableClass DataBatchUserDefinedFunction)ShuffleStrategy)EXPRESSION_API_GROUPDataset) DownloadExprExprStarExpr) PublicAPIz!Computations or Descriptive StatszFunction Applicationc!eZdZdZdedeeeeefdee fdZ defdZ e e d edefd Zd ed eeeeffd Ze e ddddddddddddd deeefdedeeefdeedeeedeeeefdeeedeeeefdeedeedeedeee ee e fee e e ffdeegeeeffddfd Ze ed!"d#ed$edefd%Ze e defd&Z!e e  d/d eeeefd'edefd(Z"e e  d/d eeeefd'edefd)Z#e e  d/d eeeefd'edefd*Z$e e  d/d eeeefd'edefd+Z%e e  d0d eeeefd-e d'edefd.Z&dS)1 GroupedDatazRepresents a grouped dataset created by calling ``Dataset.groupby()``. The actual groupby is deferred until an aggregation is applied. datasetkeynum_partitionsc0||_||_||_dS)zConstruct a dataset grouped by key (internal API). The constructor is not part of the GroupedData API. Use the ``Dataset.groupby()`` method to construct one. N)_dataset_key_num_partitions)selfr%r&r's i/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/grouped_data.py__init__zGroupedData.__init__s ") 58 .<returnc@|jjd|jd|jdS)Nz (dataset=z, key=)) __class____name__r)r*r,s r-__repr__zGroupedData.__repr__/s,~& X X X X$) X X X r/) api_groupaggsc|jj}t|jjj|j||j}t||jj }t||S)aImplements an accumulator-based aggregation. Args: aggs: Aggregations to do. Returns: The output is an dataset of ``n + 1`` columns where the first column is the groupby key and the second through ``n + 1`` columns are the results of the aggregations. If groupby key is ``None`` then the key part of return is omitted. )r&r8r') r)_plancopyr _logical_plandagr*r+rcontextr)r,r8planop logical_plans r- aggregatezGroupedData.aggregate4su}"''))  M ' + /     #2t}'<==      r/agg_clsoncR|jj||g|Rd|ji|}|j|S)aVHelper for aggregating on a particular subset of the dataset. This validates the `on` argument, and converts a list of column names to a multi-aggregation. A null `on` results in a multi-aggregation on all columns for an Arrow Dataset, and a single aggregation on the entire row for a simple Dataset. skip_cols)r)_build_multicolumn_aggsr*rB)r,rCrDargskwargsr8s r- _aggregate_onzGroupedData._aggregate_onOsT5t}4 R    *.) 7=  t~t$$r/TNdefault) zero_copy_batchcompute batch_formatfn_args fn_kwargsfn_constructor_argsfn_constructor_kwargsnum_cpusnum_gpusmemory concurrencyray_remote_args_fnfnrLrMrNrOrPrQrRrSrTrUrVrWrc  |j|jd}ny|jjjt jkr;|jp|jjj}|j||jd}n|j |j}|jgn]t|jtr |jgn:t|jtr|jntd|jdttrGfdd}nfd }ttrjj|_n j|_|j|fd|d|||||| | | | d | d |S) aApply the given function to each group of records of this dataset. While map_groups() is very flexible, note that it comes with downsides: * It may be slower than using more specific methods such as min(), max(). * It requires that each group fits in memory on a single node. In general, prefer to use `aggregate()` instead of `map_groups()`. .. warning:: Specifying both ``num_cpus`` and ``num_gpus`` for map tasks is experimental, and may result in scheduling or stability issues. Please `report any issues `_ to the Ray team. Examples: >>> # Return a single record per group (list of multiple records in, >>> # list of a single record out). >>> import ray >>> import pandas as pd >>> import numpy as np >>> # Get first value per group. >>> ds = ray.data.from_items([ # doctest: +SKIP ... {"group": 1, "value": 1}, ... {"group": 1, "value": 2}, ... {"group": 2, "value": 3}, ... {"group": 2, "value": 4}]) >>> ds.groupby("group").map_groups( # doctest: +SKIP ... lambda g: {"result": np.array([g["value"][0]])}) >>> # Return multiple records per group (dataframe in, dataframe out). >>> df = pd.DataFrame( ... {"A": ["a", "a", "b"], "B": [1, 1, 3], "C": [4, 6, 5]} ... ) >>> ds = ray.data.from_pandas(df) # doctest: +SKIP >>> grouped = ds.groupby("A") # doctest: +SKIP >>> grouped.map_groups( # doctest: +SKIP ... lambda g: g.apply( ... lambda c: c / g[c.name].sum() if c.name in ["B", "C"] else c ... ) ... ) # doctest: +SKIP Args: fn: The function to apply to each group of records, or a class type that can be instantiated to create such a callable. It takes as input a batch of all records from a single group, and returns a batch of zero or more records, similar to map_batches(). zero_copy_batch: If True, each group of rows (batch) will be provided w/o making an additional copy. compute: The compute strategy to use for the map operation. * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks. * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks. * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers. * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers. * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n, initial_size=initial)`` to use an autoscaling actor pool from ``m`` to ``n`` workers, with an initial size of ``initial``. batch_format: Specify ``"default"`` to use the default block format (NumPy), ``"pandas"`` to select ``pandas.DataFrame``, "pyarrow" to select ``pyarrow.Table``, or ``"numpy"`` to select ``Dict[str, numpy.ndarray]``, or None to return the underlying block exactly as is with no additional formatting. fn_args: Arguments to `fn`. fn_kwargs: Keyword arguments to `fn`. fn_constructor_args: Positional arguments to pass to ``fn``'s constructor. You can only provide this if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. fn_constructor_kwargs: Keyword arguments to pass to ``fn``'s constructor. This can only be provided if ``fn`` is a callable class. These arguments are top-level arguments in the underlying Ray actor construction task. num_cpus: The number of CPUs to reserve for each parallel map worker. num_gpus: The number of GPUs to reserve for each parallel map worker. For example, specify `num_gpus=1` to request 1 GPU for each parallel map worker. memory: The heap memory in bytes to reserve for each parallel map worker. ray_remote_args_fn: A function that returns a dictionary of remote args passed to each map worker. The purpose of this argument is to generate dynamic arguments for each actor or task, and will be called each time prior to initializing the worker. Args returned from this dict will always override the args in ``ray_remote_args``. Note: this is an advanced, experimental feature. concurrency: This argument is deprecated. Use ``compute`` argument. ray_remote_args: Additional resource requirements to request from Ray (e.g., num_gpus=1 to request GPUs for the map tasks). See :func:`ray.remote` for details. Returns: The return type is determined by the return type of ``fn``, and the return value is combined from results of all groups. .. seealso:: :meth:`GroupedData.aggregate` Use this method for common aggregation use cases. NT)keyssortzYGroup-by keys are expected to either be a single column (str) or a list of columns (got 'z')c&eZdZfdZfdZdS)*GroupedData.map_groups..wrapped_fnc"|i||_dSN)rX)r,rHrIrXs r-r.z3GroupedData.map_groups..wrapped_fn.__init__ s b$1&11DGGGr/c?HKt|j|g|Ri|Ed{VdSr`)_apply_udf_to_groupsrX)r,batchrHrIrNr[s r-__call__z3GroupedData.map_groups..wrapped_fn.__call__ sf3l =A   EK  r/N)r4 __module__ __qualname__r.rd)rNrXr[sr- wrapped_fnr^sL22222r/rgc?>Kt|g|Ri|Ed{VdSr`)rb)rcrHrIrNrXr[s r-rgz*GroupedData.map_groups..wrapped_fnsd/t\48shuffle_strategyr HASH_SHUFFLEr+ default_hash_shuffle_parallelismr\ isinstancestrr ValueErrorrrfuncr4*_map_batches_without_batch_size_validation)r,rXrLrMrNrOrPrQrRrSrTrUrVrWray_remote_args shuffled_dsr'rgr[s ` ` @r- map_groupszGroupedData.map_groupsbs@ 9 -33A66KK ] " 37S S S$J=(I -33Y 4KK-,,TY77K 9 DD  3 ' ' I;DD  4 ( ( 9DD<.2i<<<  b- ( (                      b' " " ."$'"2J  "$+J F{E  + 3"7#$)1%  &'   r/alpha)r7 stability column_nameexprc zt|tr|std|t|tst dt|t rt d||}t|gdtdtffd }|j |fddd |S) aAdd a new column to each group using an expression. The supplied expression is evaluated against every row in each group, and the resulting column is appended to the group's records. The output dataset preserves the original rows and columns. Examples: >>> import ray >>> from ray.data.expressions import col >>> ds = ( ... ray.data.from_items([{"group": 1, "value": 1}, {"group": 1, "value": 2}]) ... .groupby("group") ... .with_column("value_twice", col("value") * 2) ... .sort(["group", "value"]) ... ) >>> ds.take_all() [{'group': 1, 'value': 1, 'value_twice': 2}, {'group': 1, 'value': 2, 'value_twice': 4}] Args: column_name: Name of the column to add. expr: Expression that yields the values for the new column. **ray_remote_args: Additional resource requirements to request from Ray for the underlying map tasks (for example, ``num_gpus=1``). Returns: A new :class:`~ray.data.Dataset` containing all existing columns plus the newly computed column. z-column_name must be a non-empty string, got: zBexpr must be a Ray Data expression created via the expression API.zBGroupedData.with_column does not yet support download expressions.blockr0c(ddlm}||S)Nr)eval_projection)?ray.data._internal.planner.plan_expression.expression_evaluatorr~)r|r~projection_exprss r-_project_groupz/GroupedData.with_column.._project_groupjs6      #?#3U;; ;r/NT)rNrL) rorprqr TypeErrorraliasr!rrv)r,ryrzrt aliased_exprrrs @r- with_columnzGroupedData.with_column7sF+s++ ; O OO $%% T  dL ) ) T zz+.. $JJ 5 <% >> import ray >>> ray.data.from_items([ # doctest: +SKIP ... {"A": x % 3, "B": x} for x in range(100)]).groupby( # doctest: +SKIP ... "A").count() # doctest: +SKIP Returns: A dataset of ``[k, v]`` columns where ``k`` is the groupby key and ``v`` is the number of rows with that key. If groupby key is ``None`` then the key part of return is omitted. )rBrr5s r-countzGroupedData.countxs~~egg&&&r/ ignore_nullsc<|t||S)aCompute grouped sum aggregation. Examples: >>> import ray >>> ray.data.from_items([ # doctest: +SKIP ... (i % 3, i, i**2) # doctest: +SKIP ... for i in range(100)]) # doctest: +SKIP ... .groupby(lambda x: x[0] % 3) # doctest: +SKIP ... .sum(lambda x: x[2]) # doctest: +SKIP >>> ray.data.range(100).groupby("id").sum() # doctest: +SKIP >>> ray.data.from_items([ # doctest: +SKIP ... {"A": i % 3, "B": i, "C": i**2} # doctest: +SKIP ... for i in range(100)]) # doctest: +SKIP ... .groupby("A") # doctest: +SKIP ... .sum(["B", "C"]) # doctest: +SKIP Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values will be ignored when computing the sum; if ``False``, if a null value is encountered, the output will be null. We consider np.nan, None, and pd.NaT to be null values. Default is ``True``. Returns: The sum result. For different values of ``on``, the return varies: - ``on=None``: a dataset containing a groupby key column, ``"k"``, and a column-wise sum column for each original column in the dataset. - ``on=["col_1", ..., "col_n"]``: a dataset of ``n + 1`` columns where the first column is the groupby key and the second through ``n + 1`` columns are the results of the aggregations. If groupby key is ``None`` then the key part of return is omitted. r)rJrr,rDrs r-sumzGroupedData.sums T!!#r !EEEr/c<|t||S)aCompute grouped min aggregation. Examples: >>> import ray >>> ray.data.le(100).groupby("value").min() # doctest: +SKIP >>> ray.data.from_items([ # doctest: +SKIP ... {"A": i % 3, "B": i, "C": i**2} # doctest: +SKIP ... for i in range(100)]) # doctest: +SKIP ... .groupby("A") # doctest: +SKIP ... .min(["B", "C"]) # doctest: +SKIP Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values will be ignored when computing the min; if ``False``, if a null value is encountered, the output will be null. We consider np.nan, None, and pd.NaT to be null values. Default is ``True``. Returns: The min result. For different values of ``on``, the return varies: - ``on=None``: a dataset containing a groupby key column, ``"k"``, and a column-wise min column for each original column in the dataset. - ``on=["col_1", ..., "col_n"]``: a dataset of ``n + 1`` columns where the first column is the groupby key and the second through ``n + 1`` columns are the results of the aggregations. If groupby key is ``None`` then the key part of return is omitted. r)rJrrs r-minzGroupedData.min J!!#r !EEEr/c<|t||S)aCompute grouped max aggregation. Examples: >>> import ray >>> ray.data.le(100).groupby("value").max() # doctest: +SKIP >>> ray.data.from_items([ # doctest: +SKIP ... {"A": i % 3, "B": i, "C": i**2} # doctest: +SKIP ... for i in range(100)]) # doctest: +SKIP ... .groupby("A") # doctest: +SKIP ... .max(["B", "C"]) # doctest: +SKIP Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values will be ignored when computing the max; if ``False``, if a null value is encountered, the output will be null. We consider np.nan, None, and pd.NaT to be null values. Default is ``True``. Returns: The max result. For different values of ``on``, the return varies: - ``on=None``: a dataset containing a groupby key column, ``"k"``, and a column-wise max column for each original column in the dataset. - ``on=["col_1", ..., "col_n"]``: a dataset of ``n + 1`` columns where the first column is the groupby key and the second through ``n + 1`` columns are the results of the aggregations. If groupby key is ``None`` then the key part of return is omitted. r)rJrrs r-maxzGroupedData.maxrr/c<|t||S)aCompute grouped mean aggregation. Examples: >>> import ray >>> ray.data.le(100).groupby("value").mean() # doctest: +SKIP >>> ray.data.from_items([ # doctest: +SKIP ... {"A": i % 3, "B": i, "C": i**2} # doctest: +SKIP ... for i in range(100)]) # doctest: +SKIP ... .groupby("A") # doctest: +SKIP ... .mean(["B", "C"]) # doctest: +SKIP Args: on: a column name or a list of column names to aggregate. ignore_nulls: Whether to ignore null values. If ``True``, null values will be ignored when computing the mean; if ``False``, if a null value is encountered, the output will be null. We consider np.nan, None, and pd.NaT to be null values. Default is ``True``. Returns: The mean result. For different values of ``on``, the return varies: - ``on=None``: a dataset containing a groupby key column, ``"k"``, and a column-wise mean column for each original column in the dataset. - ``on=["col_1", ..., "col_n"]``: a dataset of ``n + 1`` columns where the first column is the groupby key and the second through ``n + 1`` columns are the results of the aggregations. If groupby key is ``None`` then the key part of return is omitted. r)rJrrs r-meanzGroupedData.means J!!$!FFFr/rZddofc>|t|||S)ahCompute grouped standard deviation aggregation. Examples: >>> import ray >>> ray.data.range(100).groupby("id").std(ddof=0) # doctest: +SKIP >>> ray.data.from_items([ # doctest: +SKIP ... {"A": i % 3, "B": i, "C": i**2} # doctest: +SKIP ... for i in range(100)]) # doctest: +SKIP ... .groupby("A") # doctest: +SKIP ... .std(["B", "C"]) # doctest: +SKIP NOTE: This uses Welford's online method for an accumulator-style computation of the standard deviation. This method was chosen due to it's numerical stability, and it being computable in a single pass. This may give different (but more accurate) results than NumPy, Pandas, and sklearn, which use a less numerically stable two-pass algorithm. See https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm Args: on: a column name or a list of column names to aggregate. ddof: Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. ignore_nulls: Whether to ignore null values. If ``True``, null values will be ignored when computing the std; if ``False``, if a null value is encountered, the output will be null. We consider np.nan, None, and pd.NaT to be null values. Default is ``True``. Returns: The standard deviation result. For different values of ``on``, the return varies: - ``on=None``: a dataset containing a groupby key column, ``"k"``, and a column-wise std column for each original column in the dataset. - ``on=["col_1", ..., "col_n"]``: a dataset of ``n + 1`` columns where the first column is the groupby key and the second through ``n + 1`` columns are the results of the aggregations. If groupby key is ``None`` then the key part of return is omitted. )rr)rJr)r,rDrrs r-stdzGroupedData.std*s"d!!#r 4!PPPr/)NT)NrZT)'r4rerf__doc__rr r rpr intr.r6r" FA_API_GROUPrrBtyperJrrboolr rrrfloatr rrvrr r CDS_API_GROUPrrrrrrr/r-r$r$s ==eCcN+ ,= ! ====  #    Y&&& { w   '& 4%% #tCy. !%%%%&Y&&& !%/3&/+/.27;:>$($("&SWEIR R R  9 4 5R  R sO+, R sm R (3-(R DcN+R &hsm4R  (S#X7R 5/R 5/R R eCsCx%S# :N$NOPR %Xb$sCx..@%ABR " #R R R '&R hY-AAA> > >  > > > BA> @Y''''w'''('' Y'''EI)F)FT#Y')F>B)F )F)F)F(')FVY'''EI$F$FT#Y'$F>B$F $F$F$F('$FLY'''EI$F$FT#Y'$F>B$F $F$F$F('$FLY'''EI$G$GT#Y'$G>B$G $G$G$G('$GLY'''%)! 1Q1Q #tCy. !1Q1Q 1Q  1Q1Q1Q('1Q1Q1Qr/r$udf.r|r[rNrHrIr0c/Ktj|}||}t|dd|ddD]r\}} ||| d} tj| } || |g|Ri|} t | tr | Ed{Vn| VsdS)zApply UDF to groups of rows having the same set of values of the specified columns (keys). NOTE: This function is defined at module level to avoid capturing closures and make it serializable. NrZF)r;)r for_block_get_group_boundaries_sortedzipsliceto_batch_formatro IteratorABC) rr|r[rNrHrIblock_accessor boundariesstartend group_blockgroup_block_accessorresults r-rbrb_s #,U33N<:: s$**5#E*BB ,6{CC  0 0 > > AE   IO   fk * *         LLLL#r/N)3collections.abcrr functoolsrtypingrrrrr r r r ray.data._internal.computer %ray.data._internal.logical.interfacesr8ray.data._internal.logical.operators.all_to_all_operatorrray.data.aggregaterrrrrrrray.data.blockrrrrrray.data.contextrray.data.datasetrrray.data.expressionsrr r!ray.util.annotationsr"rrr$rprbGroupedDatasetrr/r-rs`333333XXXXXXXXXXXXXXXXXXXXXX666666======NNNNNNKKKKKKKKKKKKKKKKKK-,,,,,::::::::==========******3 % C QC QC QC QC QC QC QC QL% )S!9,-)S!8I#667 9 %  % s) %3-% %%i%%%%Rr/