o bis`@sddlmZddlmZmZmZmZmZmZm Z m Z m Z ddl m Z ddlmZddlmZddlmZmZmZmZmZmZmZddlmZmZmZmZmZddl m!Z!dd l"m#Z#dd l$m%Z%d Z&d Z'Gd ddZ(deedgefdedee)de e)dededeefddZ*e(Z+dS))partial) AnyCallableDictIterableIteratorListOptionalTupleUnion)ComputeStrategy) LogicalPlan) Aggregate) AggregateFnCountMaxMeanMinStdSum)Block BlockAccessor CallableClass DataBatchUserDefinedFunction)ShuffleStrategy)Dataset) PublicAPIz!Computations or Descriptive StatszFunction Applicationc!@sjeZdZdZdedeeeeefdee fddZ defdd Z e e d d edefd d ZdedeeeeffddZe e d ddddddddddddd deeefdedeeefdeedeeedeeeefdeeedeeeefdeedeed eed!eee ee e ffd"eegeeeffdd#fd$d%Ze ed defd&d'Ze ed  (d6deeeefd)edefd*d+Ze ed  (d6deeeefd)edefd,d-Z e ed  (d6deeeefd)edefd.d/Z!e ed  (d6deeeefd)edefd0d1Z"e ed   2 (d7deeeefd3e d)edefd4d5Z#dS)8 GroupedDatazRepresents a grouped dataset created by calling ``Dataset.groupby()``. The actual groupby is deferred until an aggregation is applied. datasetkeynum_partitionscCs||_||_||_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)selfrr r!r&I/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/grouped_data.py__init__s  zGroupedData.__init__returncCs|jjd|jd|jdS)Nz (dataset=z, key=)) __class____name__r"r#r%r&r&r'__repr__-szGroupedData.__repr__) api_groupaggscGs>|jj}t|jjj|j||jd}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 r0r!) r"_plancopyr _logical_plandagr#r$r contextr)r%r0planop logical_planr&r&r' aggregate2s zGroupedData.aggregateagg_clsoncOs,|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#r9)r%r:r;argskwargsr0r&r&r' _aggregate_onMs zGroupedData._aggregate_onFNdefault) zero_copy_batchcompute batch_formatfn_args fn_kwargsfn_constructor_argsfn_constructor_kwargsnum_cpusnum_gpusmemory concurrencyray_remote_args_fnfnrBrCrDrErFrGrHrIrJrKrLrMrc  s&|jdur |jd}n"|jjjtjkr'|jp|jjj}|jj||jdd}n|j |j}|jdur6gnt |jt rA|jgnt |jt rK|jn t d|jdt treGfddd}nfd d }t trxjj|_nj|_|j|fd|d|||||| | | | | d |S) a@Apply 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: This argument is deprecated. Use ``concurrency`` argument. 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: The semantics of this argument depend on the type of ``fn``: * If ``fn`` is a function and ``concurrency`` isn't set (default), the actual concurrency is implicitly determined by the available resources and number of input blocks. * If ``fn`` is a function and ``concurrency`` is an int ``n``, Ray Data launches *at most* ``n`` concurrent tasks. * If ``fn`` is a class and ``concurrency`` is an int ``n``, Ray Data uses an actor pool with *exactly* ``n`` workers. * If ``fn`` is a class and ``concurrency`` is a tuple ``(m, n)``, Ray Data uses an autoscaling actor pool from ``m`` to ``n`` workers. * If ``fn`` is a class and ``concurrency`` isn't set (default), this method raises an error. 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')cs&eZdZfddZfddZdS)*GroupedData.map_groups..wrapped_fncs|i||_dSNrN)r%r>r?rTr&r'r( sz3GroupedData.map_groups..wrapped_fn.__init__c?s*t|j|g|Ri|EdHdSrS)_apply_udf_to_groupsrN)r%batchr>r?)rDrPr&r'__call__s z3GroupedData.map_groups..wrapped_fn.__call__N)r, __module__ __qualname__r(rWr&rDrNrPr&r' wrapped_fn s r[c?s(t|g|Ri|EdHdSrS)rU)rVr>r?rZr&r'r[srR) batch_sizerCrDrBrErFrGrHrIrJrKrLrM)r#r" repartitionr5shuffle_strategyr HASH_SHUFFLEr$ default_hash_shuffle_parallelismrQ isinstancestrr ValueErrorrrfuncr,*_map_batches_without_batch_size_validation)r%rNrBrCrDrErFrGrHrIrJrKrLrMray_remote_args shuffled_dsr!r[r&rZr' map_groups`sb        zGroupedData.map_groupscCs |tS)aCompute count aggregation. Examples: >>> 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. )r9rr-r&r&r'count8s zGroupedData.countT ignore_nullscC|jt||dS)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. rj)r@rr%r;rjr&r&r'sumIs*zGroupedData.sumcCrk)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. rl)r@rrmr&r&r'minu%zGroupedData.mincCrk)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. rl)r@rrmr&r&r'maxrpzGroupedData.maxcCrk)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. rl)r@rrmr&r&r'meanrpzGroupedData.meanrOddofcCs|jt|||dS)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. )rjrs)r@r)r%r;rsrjr&r&r'stds2zGroupedData.std)NT)NrOT)$r,rXrY__doc__rr r rbrintr(r.r FA_API_GROUPrr9typer@rrboolr rrrfloatr rrh CDS_API_GROUPrirnrorqrrrtr&r&r&r'rs           X + & & &rudf.blockrPrDr>r?r)c osvt|}||}t|dd|ddD] \}} |j|| dd} t| } || |g|Ri|VqdS)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.NrOF)r2)r for_block_get_group_boundaries_sortedzipsliceto_batch_format) r|r}rPrDr>r?block_accessor boundariesstartend group_blockgroup_block_accessorr&r&r'rUs "  rUN), functoolsrtypingrrrrrrr r r ray.data._internal.computer %ray.data._internal.logical.interfacesr 8ray.data._internal.logical.operators.all_to_all_operatorrray.data.aggregaterrrrrrrray.data.blockrrrrrray.data.contextrray.data.datasetrray.util.annotationsrr{rwrrbrUGroupedDatasetr&r&r&r'sB ,   $