code: def shuffle_group(
      df, cols, stage, k, npartitions, ignore_index, nfinal
  ):
      """Splits dataframe into groups
  
      The group is determined by their final partition, and which stage we are in
      in the shuffle
  
      Parameters
      ----------
      df: DataFrame
      cols: str or list
          Column name(s) on which to split the dataframe. If ``cols`` is not
          "_partitions", hashing will be used to determine target partition
      stage: int
          We shuffle dataframes with many partitions we in a few stages to avoid
          a quadratic number of tasks.  This number corresponds to which stage
          we're in, starting from zero up to some small integer
      k: int
          Desired number of splits from this dataframe
      npartition: int
          Total number of output partitions for the full dataframe
      nfinal: int
          Total number of output partitions after repartitioning
  
      Returns
      -------
      out: Dict[int, DataFrame]
          A dictionary mapping integers in {0..k} to dataframes such that the
          hash values of ``df[col]`` are well partitioned.
      """
      if isinstance(cols, str):
          cols = [cols]
  
      if cols and cols[0] == "_partitions":
          ind = df[cols[0]]
      else:
          ind = hash_object_dispatch(
              df[cols] if cols else df, index=False
          )
          if nfinal and nfinal != npartitions:
              ind = ind % int(nfinal)
  
      typ = np.min_scalar_type(npartitions * 2)
      # Here we convert the final output index `ind` into the output index
      # for the current stage.
      ind = (
          (ind % npartitions).astype(typ, copy=False) // k**stage % k
      )
      return group_split_dispatch(df, ind, k, ignore_index=ignore_index)
  
imports: from __future__ import annotations
  import contextlib
  import logging
  import math
  import shutil
  import tempfile
  import uuid
  import warnings
  from collections.abc import Callable, Mapping, Sequence
  from typing import Any, Literal
  import numpy
  import pandas
  import tlz
  from pandas.api.types import is_numeric_dtype
  from dask import config
  from dask.base import compute, compute_as_if_collection, is_dask_collection, tokenize
  from dask.dataframe import methods
  from dask.dataframe.core import DataFrame, Series, _deprecated_kwarg, _Frame, map_partitions, new_dd_object
  from dask.dataframe.dispatch import group_split_dispatch, hash_object_dispatch, partd_encode_dispatch
  from dask.dataframe.utils import UNKNOWN_CATEGORIES
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import ShuffleLayer, SimpleShuffleLayer
  from dask.sizeof import sizeof
  from dask.utils import M, digit, get_default_shuffle_method
label_desc: Use image node_id 18 for calling a global function with example usage: shuffle_group(df, cols, stage, k, npartitions, ignore_index, nfinal) and returns: group_split_dispatch