code: def _shuffle_aggregate(
      ddf,
      gb_cols,
      chunk,
      chunk_kwargs,
      aggregate,
      aggregate_kwargs,
      split_every,
      split_out,
      token=None,
      sort=None,
      shuffle_method=None,
  ):
      # Shuffle-based groupby aggregation
      # NOTE: This function is the dask_cudf version of
      # dask.dataframe.groupby._shuffle_aggregate
  
      # Step 1 - Chunkwise groupby operation
      chunk_name = f"{token or funcname(chunk)}-chunk"
      chunked = ddf.map_partitions(
          chunk,
          meta=chunk(ddf._meta, **chunk_kwargs),
          token=chunk_name,
          **chunk_kwargs,
      )
  
      # Step 2 - Perform global sort or shuffle
      shuffle_npartitions = max(
          chunked.npartitions // split_every,
          split_out,
      )
      if sort and split_out > 1:
          # Sort-based code path
          result = (
              chunked.repartition(npartitions=shuffle_npartitions)
              .sort_values(
                  gb_cols,
                  ignore_index=True,
                  shuffle_method=shuffle_method,
              )
              .map_partitions(
                  aggregate,
                  meta=aggregate(chunked._meta, **aggregate_kwargs),
                  **aggregate_kwargs,
              )
          )
      else:
          # Hash-based code path
          result = chunked.shuffle(
              gb_cols,
              npartitions=shuffle_npartitions,
              ignore_index=True,
              shuffle_method=shuffle_method,
          ).map_partitions(
              aggregate,
              meta=aggregate(chunked._meta, **aggregate_kwargs),
              **aggregate_kwargs,
          )
  
      # Step 3 - Repartition and return
      if split_out < result.npartitions:
          return result.repartition(npartitions=split_out)
      return result
  
imports: from functools import wraps
  from typing import Set
  import numpy
  import pandas
  from dask.dataframe.core import DataFrame, aca, split_out_on_cols
  from dask.dataframe.groupby import DataFrameGroupBy, SeriesGroupBy
  from dask.utils import funcname
  import cudf
  from cudf.utils.nvtx_annotation import _dask_cudf_nvtx_annotate
  from dask_cudf.sorting import _deprecate_shuffle_kwarg
label_desc: Use image node_id 2 for calling a global function with example usage: _shuffle_aggregate(ddf, gb_cols, chunk, chunk_kwargs, aggregate, aggregate_kwargs, split_every, split_out, token, sort, shuffle_method) and returns: result, result