Repository: dask/dask

Issue:
Deprecate ``shuffle`` keyword in favour of ``shuffle_method`` for DataFrame methods
dask-expr already changed the shuffle keyword to be a bit more explicit, we should deprecate this here to enforce when moving over

Get us consistent with the config option as well

Discussion:
phofl: @hendrikmakait what's the status here?
hendrikmakait: I'm finishing up the sibling PR over at `dask/distributed` to get ahead of a failure cascade.

Patch:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 7cddc42d9..e1e7904a4 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -84,6 +84,7 @@ from dask.utils import (
     M,
     OperatorMethodMixin,
     _deprecated,
+    _deprecated_kwarg,
     apply,
     derived_from,
     funcname,
@@ -870,7 +871,7 @@ Dask Name: {name}, {layers}"""
             raise ValueError(msg)
 
     def _drop_duplicates_shuffle(
-        self, split_out, split_every, shuffle, ignore_index, **kwargs
+        self, split_out, split_every, shuffle_method, ignore_index, **kwargs
     ):
         # Private method that drops duplicate rows using a
         # shuffle-based algorithm.
@@ -907,7 +908,7 @@ Dask Name: {name}, {layers}"""
                 kwargs.get("subset", None) or list(df.columns),
                 ignore_index=ignore_index,
                 npartitions=shuffle_npartitions,
-                shuffle=shuffle,
+                shuffle_method=shuffle_method,
             )
             .map_partitions(
                 chunk,
@@ -1758,12 +1759,13 @@ Dask Name: {name}, {layers}"""
         elif freq is not None:
             return repartition_freq(self, freq=freq)
 
+    @_deprecated_kwarg("shuffle", "shuffle_method")
     def shuffle(
         self,
         on,
         npartitions=None,
         max_branch=None,
-        shuffle=None,
+        shuffle_method=None,
         ignore_index=False,
         compute=None,
     ):
@@ -1783,7 +1785,7 @@ Dask Name: {name}, {layers}"""
         max_branch: int, optional
             The maximum number of splits per input partition. Used within
             the staged shuffling algorithm.
-        shuffle: {'disk', 'tasks', 'p2p'}, optional
+        shuffle_method: {'disk', 'tasks', 'p2p'}, optional
             Either ``'disk'`` for single-node operation or ``'tasks'`` and
             ``'p2p'`` for distributed operation.  Will be inferred by your
             current scheduler.
@@ -1809,7 +1811,7 @@ Dask Name: {name}, {layers}"""
             on,
             npartitions=npartitions,
             max_branch=max_branch,
-            shuffle=shuffle,
+            shuffle_method=shuffle_method,
             ignore_index=ignore_index,
             compute=compute,
         )
@@ -3735,6 +3737,12 @@ Dask Name: {name}, {layers}"""
 
         return Resampler(self, rule, closed=closed, label=label)
 
+    @_deprecated(
+        message=(
+            "Will be removed in a future version. "
+            "Please create a mask and filter using .loc instead"
+        )
+    )
     @derived_from(pd.DataFrame)
     def first(self, offset):
         # Let pandas error on bad args
@@ -3769,6 +3777,12 @@ Dask Name: {name}, {layers}"""
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=[self])
         return new_dd_object(graph, name, self, divs)
 
+    @_deprecated(
+        message=(
+            "Will be removed in a future version. "
+            "Please create a mask and filter using .loc instead"
+        )
+    )
     @derived_from(pd.DataFrame)
     def last(self, offset):
         # Let pandas error on bad args
@@ -5326,7 +5340,7 @@ class DataFrame(_Frame):
             If ``True``, sort the DataFrame by the new index. Otherwise
             set the index on the individual existing partitions.
             Defaults to ``True``.
-        shuffle: {'disk', 'tasks', 'p2p'}, optional
+        shuffle_method: {'disk', 'tasks', 'p2p'}, optional
             Either ``'disk'`` for single-node operation or ``'tasks'`` and
             ``'p2p'`` for distributed operation.  Will be inferred by your
             current scheduler.
@@ -5797,6 +5811,7 @@ class DataFrame(_Frame):
             "Drop currently only works for axis=1 or when columns is not None"
         )
 
+    @_deprecated_kwarg("shuffle", "shuffle_method")
     def merge(
         self,
         right,
@@ -5809,7 +5824,7 @@ class DataFrame(_Frame):
         suffixes=("_x", "_y"),
         indicator=False,
         npartitions=None,
-        shuffle=None,
+        shuffle_method=None,
         broadcast=None,
     ):
         """Merge the DataFrame with another DataFrame
@@ -5863,7 +5878,7 @@ class DataFrame(_Frame):
             performing a hash_join (merging on columns only). If ``None`` then
             ``npartitions = max(lhs.npartitions, rhs.npartitions)``.
             Default is ``None``.
-        shuffle: {'disk', 'tasks', 'p2p'}, optional
+        shuffle_method: {'disk', 'tasks', 'p2p'}, optional
             Either ``'disk'`` for single-node operation or ``'tasks'`` and
             ``'p2p'``` for distributed operation.  Will be inferred by your
             current scheduler.
@@ -5919,10 +5934,11 @@ class DataFrame(_Frame):
             suffixes=suffixes,
             npartitions=npartitions,
             indicator=indicator,
-            shuffle=shuffle,
+            shuffle_method=shuffle_method,
             broadcast=broadcast,
         )
 
+    @_deprecated_kwarg("shuffle", "shuffle_method")
     @derived_from(pd.DataFrame)
     def join(
         self,
@@ -5932,7 +5948,7 @@ class DataFrame(_Frame):
         lsuffix="",
         rsuffix="",
         npartitions=None,
-        shuffle=None,
+        shuffle_method=None,
     ):
         if is_series_like(other) and hasattr(other, "name"):
             other = other.to_frame()
@@ -5957,7 +5973,7 @@ class DataFrame(_Frame):
                     lsuffix=lsuffix,
                     rsuffix=rsuffix,
                     npartitions=npartitions,
-                    shuffle=shuffle,
+                    shuffle_method=shuffle_method,
                 )
             else:
                 # Do recursive pairwise join on everything _except_ the last join
@@ -5968,7 +5984,7 @@ class DataFrame(_Frame):
                     lsuffix=lsuffix,
                     rsuffix=rsuffix,
                     npartitions=npartitions,
-                    shuffle=shuffle,
+                    shuffle_method=shuffle_method,
                 )
 
         from dask.dataframe.multi import merge
@@ -5982,7 +5998,7 @@ class DataFrame(_Frame):
             left_on=on,
             suffixes=(lsuffix, rsuffix),
             npartitions=npartitions,
-            shuffle=shuffle,
+            shuffle_method=shuffle_method,
         )
 
     if not PANDAS_GE_200:
diff --git a/dask/dataframe/groupby.py b/dask/dataframe/groupby.py
index 4d39d0574..c1b808603 100644
--- a/dask/dataframe/groupby.py
+++ b/dask/dataframe/groupby.py
@@ -1579,7 +1579,7 @@ class _GroupBy:
                 token=token,
                 split_every=split_every,
                 split_out=split_out,
-                shuffle=shuffle,
+                shuffle_method=shuffle,
                 sort=self.sort,
             )
 
@@ -1903,6 +1903,11 @@ class _GroupBy:
         skipna=True,
         numeric_only=no_default,
     ):
+        if axis != no_default:
+            warnings.warn(
+                "`axis` parameter is deprecated and will be removed in a future version.",
+                FutureWarning,
+            )
         if axis in (1, "columns"):
             raise NotImplementedError(
                 f"The axis={axis} keyword is not implemented for groupby.idxmin"
@@ -1933,6 +1938,12 @@ class _GroupBy:
         skipna=True,
         numeric_only=no_default,
     ):
+        if axis != no_default:
+            warnings.warn(
+                "`axis` parameter is deprecated and will be removed in a future version.",
+                FutureWarning,
+            )
+
         if axis in (1, "columns"):
             raise NotImplementedError(
                 f"The axis={axis} keyword is not implemented for groupby.idxmax"
@@ -2017,7 +2028,7 @@ class _GroupBy:
             },
             split_every=split_every,
             split_out=split_out,
-            shuffle=shuffle,
+            shuffle_method=shuffle,
             sort=self.sort,
         )
 
@@ -2347,7 +2358,7 @@ class _GroupBy:
                     token="aggregate",
                     split_every=split_every,
                     split_out=split_out,
-                    shuffle=shuffle,
+                    shuffle_method=shuffle,
                     sort=self.sort,
                 )
             else:
@@ -2371,7 +2382,7 @@ class _GroupBy:
                     token="aggregate",
                     split_every=split_every,
                     split_out=split_out,
-                    shuffle=shuffle,
+                    shuffle_method=shuffle,
                     sort=self.sort,
                 )
         else:
@@ -2640,6 +2651,11 @@ class _GroupBy:
         >>> ddf = dask.datasets.timeseries(freq="1h")
         >>> result = ddf.groupby("name").shift(1, meta={"id": int, "x": float, "y": float})
         """
+        if axis != no_default:
+            warnings.warn(
+                "`axis` parameter is deprecated and will be removed in a future version.",
+                FutureWarning,
+            )
         axis = self._normalize_axis(axis, "shift")
         kwargs = {"periods": periods, "axis": axis}
         if freq is not no_default:
@@ -2776,8 +2792,16 @@ class _GroupBy:
         if axis is no_default:
             axis = 0
 
+        if axis in ("index", 1):
+            warnings.warn(
+                "Using axis=1 in GroupBy does not require grouping and will be removed "
+                "entirely in a future version.",
+                FutureWarning,
+            )
+
         return axis
 
+    @_deprecated(message="Please use `ffill`/`bfill` or `fillna` without a GroupBy.")
     def fillna(self, value=None, method=None, limit=None, axis=no_default):
         """Fill NA/NaN values using the specified method.
 
@@ -3169,7 +3193,7 @@ def _shuffle_aggregate(
     split_out=1,
     sort=True,
     ignore_index=False,
-    shuffle="tasks",
+    shuffle_method="tasks",
 ):
     """Shuffle-based groupby aggregation
 
@@ -3205,8 +3229,8 @@ def _shuffle_aggregate(
         Whether the index can be ignored during the shuffle.
     sort : bool
         If allowed, sort the keys of the output aggregation.
-    shuffle : str, default "tasks"
-        Shuffle option to be used by ``DataFrame.shuffle``.
+    shuffle_method : str, default "tasks"
+        Shuffle method to be used by ``DataFrame.shuffle``.
     """
 
     if chunk_kwargs is None:
@@ -3285,7 +3309,7 @@ def _shuffle_aggregate(
             result = chunked.sort_values(
                 index_cols,
                 npartitions=shuffle_npartitions,
-                shuffle=shuffle,
+                shuffle_method=shuffle_method,
             ).map_partitions(
                 M.set_index,
                 index_cols,
@@ -3296,14 +3320,14 @@ def _shuffle_aggregate(
             result = chunked.set_index(
                 index_cols,
                 npartitions=shuffle_npartitions,
-                shuffle=shuffle,
+                shuffle_method=shuffle_method,
             )
     else:
         result = chunked.shuffle(
             chunked.index,
             ignore_index=ignore_index,
             npartitions=shuffle_npartitions,
-            shuffle=shuffle,
+            shuffle_method=shuffle_method,
         )
 
     # Aggregate
diff --git a/dask/dataframe/multi.py b/dask/dataframe/multi.py
index a46530fd2..b82fd2558 100644
--- a/dask/dataframe/multi.py
+++ b/dask/dataframe/multi.py
@@ -72,6 +72,7 @@ from dask.dataframe.core import (
     Index,
     Series,
     _concat,
+    _deprecated_kwarg,
     _Frame,
     _maybe_from_pandas,
     is_broadcastable,
@@ -327,6 +328,7 @@ def merge_indexed_dataframes(lhs, rhs, left_index=True, right_index=True, **kwar
 shuffle_func = shuffle  # name sometimes conflicts with keyword argument
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def hash_join(
     lhs,
     left_on,
@@ -335,7 +337,7 @@ def hash_join(
     how="inner",
     npartitions=None,
     suffixes=("_x", "_y"),
-    shuffle=None,
+    shuffle_method=None,
     indicator=False,
     max_branch=None,
 ):
@@ -346,9 +348,9 @@ def hash_join(
 
     >>> hash_join(lhs, 'id', rhs, 'id', how='left', npartitions=10)  # doctest: +SKIP
     """
-    if shuffle is None:
-        shuffle = get_default_shuffle_method()
-    if shuffle == "p2p":
+    if shuffle_method is None:
+        shuffle_method = get_default_shuffle_method()
+    if shuffle_method == "p2p":
         from distributed.shuffle import hash_join_p2p
 
         return hash_join_p2p(
@@ -365,10 +367,18 @@ def hash_join(
         npartitions = max(lhs.npartitions, rhs.npartitions)
 
     lhs2 = shuffle_func(
-        lhs, left_on, npartitions=npartitions, shuffle=shuffle, max_branch=max_branch
+        lhs,
+        left_on,
+        npartitions=npartitions,
+        shuffle_method=shuffle_method,
+        max_branch=max_branch,
     )
     rhs2 = shuffle_func(
-        rhs, right_on, npartitions=npartitions, shuffle=shuffle, max_branch=max_branch
+        rhs,
+        right_on,
+        npartitions=npartitions,
+        shuffle_method=shuffle_method,
+        max_branch=max_branch,
     )
 
     if isinstance(left_on, Index):
@@ -508,6 +518,7 @@ def warn_dtype_mismatch(left, right, left_on, right_on):
             )
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 @wraps(pd.merge)
 def merge(
     left,
@@ -521,7 +532,7 @@ def merge(
     suffixes=("_x", "_y"),
     indicator=False,
     npartitions=None,
-    shuffle=None,
+    shuffle_method=None,
     max_branch=None,
     broadcast=None,
 ):
@@ -645,12 +656,20 @@ def merge(
 
         if merge_indexed_left and left.known_divisions:
             right = rearrange_by_divisions(
-                right, right_on, left.divisions, max_branch, shuffle=shuffle
+                right,
+                right_on,
+                left.divisions,
+                max_branch,
+                shuffle_method=shuffle_method,
             )
             left = left.clear_divisions()
         elif merge_indexed_right and right.known_divisions:
             left = rearrange_by_divisions(
-                left, left_on, right.divisions, max_branch, shuffle=shuffle
+                left,
+                left_on,
+                right.divisions,
+                max_branch,
+                shuffle_method=shuffle_method,
             )
             right = right.clear_divisions()
         return map_partitions(
@@ -690,7 +709,7 @@ def merge(
         n_small = min(left.npartitions, right.npartitions)
         n_big = max(left.npartitions, right.npartitions)
         if (
-            shuffle in ("tasks", "p2p", None)
+            shuffle_method in ("tasks", "p2p", None)
             and how in ("inner", "left", "right")
             and how != bcast_side
             and broadcast is not False
@@ -731,7 +750,7 @@ def merge(
             how,
             npartitions,
             suffixes,
-            shuffle=shuffle,
+            shuffle_method=shuffle_method,
             indicator=indicator,
             max_branch=max_branch,
         )
@@ -1450,6 +1469,7 @@ def _merge_chunk_wrapper(*args, **kwargs):
     )
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def broadcast_join(
     lhs,
     left_on,
@@ -1458,7 +1478,7 @@ def broadcast_join(
     how="inner",
     npartitions=None,
     suffixes=("_x", "_y"),
-    shuffle=None,
+    shuffle_method=None,
     indicator=False,
     parts_out=None,
 ):
@@ -1506,7 +1526,7 @@ def broadcast_join(
             lhs2 = shuffle_func(
                 lhs,
                 left_on,
-                shuffle="tasks",
+                shuffle_method="tasks",
             )
             lhs_name = lhs2._name
             lhs_dep = lhs2
@@ -1516,7 +1536,7 @@ def broadcast_join(
             rhs2 = shuffle_func(
                 rhs,
                 right_on,
-                shuffle="tasks",
+                shuffle_method="tasks",
             )
             lhs_name = lhs._name
             lhs_dep = lhs
@@ -1593,7 +1613,7 @@ def broadcast_join(
 
 
 def _recursive_pairwise_outer_join(
-    dataframes_to_merge, on, lsuffix, rsuffix, npartitions, shuffle
+    dataframes_to_merge, on, lsuffix, rsuffix, npartitions, shuffle_method
 ):
     """
     Schedule the merging of a list of dataframes in a pairwise method. This is a recursive function that results
@@ -1613,7 +1633,7 @@ def _recursive_pairwise_outer_join(
         "lsuffix": lsuffix,
         "rsuffix": rsuffix,
         "npartitions": npartitions,
-        "shuffle": shuffle,
+        "shuffle_method": shuffle_method,
     }
 
     # Base case 1: just return the provided dataframe and merge with `left`
diff --git a/dask/dataframe/rolling.py b/dask/dataframe/rolling.py
index 09a412e77..799d2ad0b 100644
--- a/dask/dataframe/rolling.py
+++ b/dask/dataframe/rolling.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import datetime
+import warnings
 from numbers import Integral
 
 import pandas as pd
@@ -479,6 +480,13 @@ class Rolling:
         # See https://github.com/pandas-dev/pandas/issues/15969
         self._win_type = None if isinstance(self.window, int) else "freq"
 
+        if self.axis in ("index", 1, "rows"):
+            warnings.warn(
+                "Using axis=1 in Rolling has been deprecated and will be removed "
+                "in a future version.",
+                FutureWarning,
+            )
+
     def _rolling_kwargs(self):
         kwargs = {
             "window": self.window,
diff --git a/dask/dataframe/shuffle.py b/dask/dataframe/shuffle.py
index d99320828..270df6da4 100644
--- a/dask/dataframe/shuffle.py
+++ b/dask/dataframe/shuffle.py
@@ -18,7 +18,14 @@ 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, _Frame, map_partitions, new_dd_object
+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,
@@ -120,11 +127,12 @@ def _calculate_divisions(
     return divisions, mins.tolist(), maxes.tolist(), presorted
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def sort_values(
     df: DataFrame,
     by: str | list[str],
     npartitions: int | Literal["auto"] | None = None,
-    shuffle: str | None = None,
+    shuffle_method: str | None = None,
     ascending: bool | list[bool] = True,
     na_position: Literal["first"] | Literal["last"] = "last",
     upsample: float = 1.0,
@@ -197,7 +205,7 @@ def sort_values(
         df,
         by[0],
         divisions,
-        shuffle=shuffle,
+        shuffle_method=shuffle_method,
         ascending=ascending,
         na_position=na_position,
         duplicates=False,
@@ -206,11 +214,12 @@ def sort_values(
     return df
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def set_index(
     df: DataFrame,
     index: str | Series,
     npartitions: int | Literal["auto"] | None = None,
-    shuffle: str | None = None,
+    shuffle_method: str | None = None,
     compute: bool = False,
     drop: bool = True,
     upsample: float = 1.0,
@@ -249,17 +258,24 @@ def set_index(
             return result.map_partitions(M.sort_index)
 
     return set_partition(
-        df, index, divisions, shuffle=shuffle, drop=drop, compute=compute, **kwargs
+        df,
+        index,
+        divisions,
+        shuffle_method=shuffle_method,
+        drop=drop,
+        compute=compute,
+        **kwargs,
     )
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def set_partition(
     df: DataFrame,
     index: str | Series,
     divisions: Sequence,
     max_branch: int = 32,
     drop: bool = True,
-    shuffle: str | None = None,
+    shuffle_method: str | None = None,
     compute: bool | None = None,
 ) -> DataFrame:
     """Group DataFrame by index
@@ -278,7 +294,7 @@ def set_partition(
         Values to form new divisions between partitions
     drop: bool, default True
         Whether to delete columns to be used as the new index
-    shuffle: str (optional)
+    shuffle_method: str (optional)
         Either 'disk' for an on-disk shuffle or 'tasks' to use the task
         scheduling framework.  Use 'disk' if you are on a single machine
         and 'tasks' if you are on a distributed cluster.
@@ -335,7 +351,7 @@ def set_partition(
         "_partitions",
         max_branch=max_branch,
         npartitions=len(divisions) - 1,
-        shuffle=shuffle,
+        shuffle_method=shuffle_method,
         compute=compute,
         ignore_index=True,
     )
@@ -362,10 +378,11 @@ def set_partition(
     return df4.map_partitions(M.sort_index)
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def shuffle(
     df,
     index,
-    shuffle=None,
+    shuffle_method=None,
     npartitions=None,
     max_branch=32,
     ignore_index=False,
@@ -388,8 +405,8 @@ def shuffle(
     shuffle_disk
     """
     list_like = pd.api.types.is_list_like(index) and not is_dask_collection(index)
-    shuffle = shuffle or get_default_shuffle_method()
-    if shuffle == "tasks" and (isinstance(index, str) or list_like):
+    shuffle_method = shuffle_method or get_default_shuffle_method()
+    if shuffle_method == "tasks" and (isinstance(index, str) or list_like):
         # Avoid creating the "_partitions" column if possible.
         # We currently do this if the user is passing in
         # specific column names (and shuffle == "tasks").
@@ -404,7 +421,7 @@ def shuffle(
                 index,
                 npartitions=npartitions,
                 max_branch=max_branch,
-                shuffle=shuffle,
+                shuffle_method=shuffle_method,
                 ignore_index=ignore_index,
                 compute=compute,
             )
@@ -445,7 +462,7 @@ def shuffle(
         "_partitions",
         npartitions=npartitions,
         max_branch=max_branch,
-        shuffle=shuffle,
+        shuffle_method=shuffle_method,
         compute=compute,
         ignore_index=ignore_index,
     )
@@ -453,12 +470,13 @@ def shuffle(
     return df3
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def rearrange_by_divisions(
     df,
     column,
     divisions,
     max_branch=None,
-    shuffle=None,
+    shuffle_method=None,
     ascending=True,
     na_position="last",
     duplicates=True,
@@ -488,45 +506,50 @@ def rearrange_by_divisions(
         "_partitions",
         max_branch=max_branch,
         npartitions=len(divisions) - 1,
-        shuffle=shuffle,
+        shuffle_method=shuffle_method,
     )
     del df3["_partitions"]
     return df3
 
 
+@_deprecated_kwarg("shuffle", "shuffle_method")
 def rearrange_by_column(
     df,
     col,
     npartitions=None,
     max_branch=None,
-    shuffle=None,
+    shuffle_method=None,
     compute=None,
     ignore_index=False,
 ):
-    shuffle = shuffle or get_default_shuffle_method()
+    shuffle_method = shuffle_method or get_default_shuffle_method()
 
     # if the requested output partitions < input partitions
     # we repartition first as shuffling overhead is
     # proportionate to the number of input partitions
 
-    if shuffle != "p2p" and npartitions is not None and npartitions < df.npartitions:
+    if (
+        shuffle_method != "p2p"
+        and npartitions is not None
+        and npartitions < df.npartitions
+    ):
         df = df.repartition(npartitions=npartitions)
 
-    if shuffle == "disk":
+    if shuffle_method == "disk":
         return rearrange_by_column_disk(df, col, npartitions, compute=compute)
-    elif shuffle == "tasks":
+    elif shuffle_method == "tasks":
         df2 = rearrange_by_column_tasks(
             df, col, max_branch, npartitions, ignore_index=ignore_index
         )
         if ignore_index:
             df2._meta = df2._meta.reset_index(drop=True)
         return df2
-    elif shuffle == "p2p":
+    elif shuffle_method == "p2p":
         from distributed.shuffle import rearrange_by_column_p2p
 
         return rearrange_by_column_p2p(df, col, npartitions)
     else:
-        raise NotImplementedError("Unknown shuffle method %s" % shuffle)
+        raise NotImplementedError("Unknown shuffle method %s" % shuffle_method)
 
 
 class maybe_buffered_partd:
diff --git a/dask/utils.py b/dask/utils.py
index e2ebb6a58..41aa08b18 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -15,12 +15,12 @@ from collections.abc import Hashable, Iterable, Iterator, Mapping, Set
 from contextlib import contextmanager, nullcontext, suppress
 from datetime import datetime, timedelta
 from errno import ENOENT
-from functools import lru_cache
+from functools import lru_cache, wraps
 from importlib import import_module
 from numbers import Integral, Number
 from operator import add
 from threading import Lock
-from typing import Any, ClassVar, Literal, TypeVar, overload
+from typing import Any, Callable, ClassVar, Literal, TypeVar, cast, overload
 from weakref import WeakValueDictionary
 
 import tlz as toolz
@@ -32,6 +32,10 @@ K = TypeVar("K")
 V = TypeVar("V")
 T = TypeVar("T")
 
+# used in decorators to preserve the signature of the function it decorates
+# see https://mypy.readthedocs.io/en/stable/generics.html#declaring-decorators
+FuncType = Callable[..., Any]
+F = TypeVar("F", bound=FuncType)
 
 system_encoding = sys.getdefaultencoding()
 if system_encoding == "ascii":
@@ -138,6 +142,126 @@ def _deprecated(
     return decorator
 
 
+def _deprecated_kwarg(
+    old_arg_name: str,
+    new_arg_name: str | None,
+    mapping: Mapping[Any, Any] | Callable[[Any], Any] | None = None,
+    stacklevel: int = 2,
+) -> Callable[[F], F]:
+    """
+    Decorator to deprecate a keyword argument of a function.
+
+    Parameters
+    ----------
+    old_arg_name : str
+        Name of argument in function to deprecate
+    new_arg_name : str or None
+        Name of preferred argument in function. Use None to raise warning that
+        ``old_arg_name`` keyword is deprecated.
+    mapping : dict or callable
+        If mapping is present, use it to translate old arguments to
+        new arguments. A callable must do its own value checking;
+        values not found in a dict will be forwarded unchanged.
+
+    Examples
+    --------
+    The following deprecates 'cols', using 'columns' instead
+
+    >>> @_deprecated_kwarg(old_arg_name='cols', new_arg_name='columns')
+    ... def f(columns=''):
+    ...     print(columns)
+    ...
+    >>> f(columns='should work ok')
+    should work ok
+
+    >>> f(cols='should raise warning')  # doctest: +SKIP
+    FutureWarning: cols is deprecated, use columns instead
+      warnings.warn(msg, FutureWarning)
+    should raise warning
+
+    >>> f(cols='should error', columns="can\'t pass do both")  # doctest: +SKIP
+    TypeError: Can only specify 'cols' or 'columns', not both
+
+    >>> @_deprecated_kwarg('old', 'new', {'yes': True, 'no': False})
+    ... def f(new=False):
+    ...     print('yes!' if new else 'no!')
+    ...
+    >>> f(old='yes')  # doctest: +SKIP
+    FutureWarning: old='yes' is deprecated, use new=True instead
+      warnings.warn(msg, FutureWarning)
+    yes!
+
+    To raise a warning that a keyword will be removed entirely in the future
+
+    >>> @_deprecated_kwarg(old_arg_name='cols', new_arg_name=None)
+    ... def f(cols='', another_param=''):
+    ...     print(cols)
+    ...
+    >>> f(cols='should raise warning')  # doctest: +SKIP
+    FutureWarning: the 'cols' keyword is deprecated and will be removed in a
+    future version please takes steps to stop use of 'cols'
+    should raise warning
+    >>> f(another_param='should not raise warning')  # doctest: +SKIP
+    should not raise warning
+
+    >>> f(cols='should raise warning', another_param='')  # doctest: +SKIP
+    FutureWarning: the 'cols' keyword is deprecated and will be removed in a
+    future version please takes steps to stop use of 'cols'
+    should raise warning
+    """
+    if mapping is not None and not hasattr(mapping, "get") and not callable(mapping):
+        raise TypeError(
+            "mapping from old to new argument values must be dict or callable!"
+        )
+
+    def _deprecated_kwarg(func: F) -> F:
+        @wraps(func)
+        def wrapper(*args, **kwargs) -> Callable[..., Any]:
+            old_arg_value = kwargs.pop(old_arg_name, None)
+
+            if old_arg_value is not None:
+                if new_arg_name is None:
+                    msg = (
+                        f"the {repr(old_arg_name)} keyword is deprecated and "
+                        "will be removed in a future version. Please take "
+                        f"steps to stop the use of {repr(old_arg_name)}"
+                    )
+                    warnings.warn(msg, FutureWarning, stacklevel=stacklevel)
+                    kwargs[old_arg_name] = old_arg_value
+                    return func(*args, **kwargs)
+
+                elif mapping is not None:
+                    if callable(mapping):
+                        new_arg_value = mapping(old_arg_value)
+                    else:
+                        new_arg_value = mapping.get(old_arg_value, old_arg_value)
+                    msg = (
+                        f"the {old_arg_name}={repr(old_arg_value)} keyword is "
+                        "deprecated, use "
+                        f"{new_arg_name}={repr(new_arg_value)} instead."
+                    )
+                else:
+                    new_arg_value = old_arg_value
+                    msg = (
+                        f"the {repr(old_arg_name)} keyword is deprecated, "
+                        f"use {repr(new_arg_name)} instead."
+                    )
+
+                warnings.warn(msg, FutureWarning, stacklevel=stacklevel)
+                if kwargs.get(new_arg_name) is not None:
+                    msg = (
+                        f"Can only specify {repr(old_arg_name)} "
+                        f"or {repr(new_arg_name)}, not both."
+                    )
+                    raise TypeError(msg)
+                kwargs[new_arg_name] = new_arg_value
+            return func(*args, **kwargs)
+
+        return cast(F, wrapper)
+
+    return _deprecated_kwarg
+
+
 def deepmap(func, *seqs):
     """Apply function inside nested lists
 


Test patch:
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 364e385eb..495251608 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -4676,13 +4676,22 @@ def test_first_and_last(method):
         for offset in offsets:
             with _check_warning(PANDAS_GE_210, FutureWarning, method):
                 expected = f(df, offset)
-            with _check_warning(PANDAS_GE_210, FutureWarning, method):
+
+            ctx = (
+                pytest.warns(
+                    FutureWarning, match="Will be removed in a future version."
+                )
+                if not PANDAS_GE_210
+                else contextlib.nullcontext()
+            )
+
+            with _check_warning(PANDAS_GE_210, FutureWarning, method), ctx:
                 actual = f(ddf, offset)
             assert_eq(actual, expected)
 
             with _check_warning(PANDAS_GE_210, FutureWarning, method):
                 expected = f(df.A, offset)
-            with _check_warning(PANDAS_GE_210, FutureWarning, method):
+            with _check_warning(PANDAS_GE_210, FutureWarning, method), ctx:
                 actual = f(ddf.A, offset)
             assert_eq(actual, expected)
 
@@ -5735,7 +5744,7 @@ def test_dask_layers():
 
 def test_repr_html_dataframe_highlevelgraph():
     pytest.importorskip("jinja2")
-    x = timeseries().shuffle("id", shuffle="tasks").head(compute=False)
+    x = timeseries().shuffle("id", shuffle_method="tasks").head(compute=False)
     hg = x.dask
     assert xml.etree.ElementTree.fromstring(hg._repr_html_()) is not None
     for layer in hg.layers.values():
@@ -5998,7 +6007,7 @@ def test_empty():
 def test_repr_materialize():
     # DataFrame/Series repr should not materialize
     # any layers in timeseries->shuffle->getitem
-    s = timeseries(end="2000-01-03").shuffle("id", shuffle="tasks")["id"]
+    s = timeseries(end="2000-01-03").shuffle("id", shuffle_method="tasks")["id"]
     assert all([not l.is_materialized() for l in s.dask.layers.values()])
     s.__repr__()
     s.to_frame().__repr__()
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 4d604b92d..c47650ddf 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -107,11 +107,12 @@ def auto_shuffle_method(shuffle_method):
 
 
 @contextlib.contextmanager
-def groupby_axis_deprecated():
-    ctx = contextlib.nullcontext()
-    if PANDAS_GE_210:
-        ctx = pytest.warns(FutureWarning, match="axis")
-    with ctx:
+def groupby_axis_deprecated(*contexts):
+    with contextlib.ExitStack() as stack:
+        for ctx in contexts:
+            stack.enter_context(ctx)
+        if PANDAS_GE_210:
+            stack.enter_context(pytest.warns(FutureWarning, match="axis"))
         yield
 
 
@@ -1311,27 +1312,36 @@ def test_fillna(axis, group_keys, limit):
     )
     ddf = dd.from_pandas(df, npartitions=2)
 
+    deprecated_ctx = pytest.warns(
+        FutureWarning, match="Please use `ffill`/`bfill` or `fillna` without a GroupBy"
+    )
+
     with groupby_axis_deprecated():
         expected = df.groupby("A", group_keys=group_keys).fillna(0, axis=axis)
-    with groupby_axis_deprecated():
+    with groupby_axis_deprecated(
+        contextlib.nullcontext() if PANDAS_GE_210 else deprecated_ctx
+    ):
         result = ddf.groupby("A", group_keys=group_keys).fillna(0, axis=axis)
     assert_eq(expected, result)
-    assert_eq(
-        df.groupby("A", group_keys=group_keys).B.fillna(0),
-        ddf.groupby("A", group_keys=group_keys).B.fillna(0),
-    )
-    assert_eq(
-        df.groupby(["A", "B"], group_keys=group_keys).fillna(0),
-        ddf.groupby(["A", "B"], group_keys=group_keys).fillna(0),
-    )
 
-    with pytest.raises(NotImplementedError):
+    with deprecated_ctx:
+        assert_eq(
+            df.groupby("A", group_keys=group_keys).B.fillna(0),
+            ddf.groupby("A", group_keys=group_keys).B.fillna(0),
+        )
+    with deprecated_ctx:
+        assert_eq(
+            df.groupby(["A", "B"], group_keys=group_keys).fillna(0),
+            ddf.groupby(["A", "B"], group_keys=group_keys).fillna(0),
+        )
+
+    with pytest.raises(NotImplementedError), deprecated_ctx:
         ddf.groupby("A").fillna({"A": 0})
 
-    with pytest.raises(NotImplementedError):
+    with pytest.raises(NotImplementedError), deprecated_ctx:
         ddf.groupby("A").fillna(pd.Series(dtype=int))
 
-    with pytest.raises(NotImplementedError):
+    with pytest.raises(NotImplementedError), deprecated_ctx:
         ddf.groupby("A").fillna(pd.DataFrame)
 
 
@@ -1757,13 +1767,19 @@ def test_cumulative_axis(func):
     assert_eq(expected, result)
 
     # axis=1
+    deprecate_ctx = pytest.warns(
+        FutureWarning,
+        match="Using axis=1 in GroupBy does not require grouping and will be removed entirely in a future version",
+    )
     with groupby_axis_deprecated():
         expected = getattr(df.groupby("a"), func)(axis=1)
-    with groupby_axis_deprecated():
+    with groupby_axis_deprecated(
+        contextlib.nullcontext() if PANDAS_GE_210 else deprecate_ctx
+    ):
         result = getattr(ddf.groupby("a"), func)(axis=1)
     assert_eq(expected, result)
 
-    with groupby_axis_deprecated():
+    with deprecate_ctx:
         with pytest.raises(ValueError, match="No axis named 1 for object type Series"):
             getattr(ddf.groupby("a").b, func)(axis=1)
 
@@ -2362,13 +2378,21 @@ def test_df_groupby_idx_axis(func, axis):
     ).set_index("idx")
 
     ddf = dd.from_pandas(pdf, npartitions=2)
+
     if axis in (1, "columns"):
-        with pytest.raises(NotImplementedError):
+        with pytest.raises(NotImplementedError), pytest.warns(
+            FutureWarning, match="`axis` parameter is deprecated"
+        ):
             getattr(ddf.groupby("group"), func)(axis=axis)
     else:
         with groupby_axis_deprecated():
             expected = getattr(pdf.groupby("group"), func)(axis=axis)
-        with groupby_axis_deprecated():
+        deprecate_ctx = pytest.warns(
+            FutureWarning, match="`axis` parameter is deprecated"
+        )
+        with groupby_axis_deprecated(
+            contextlib.nullcontext() if PANDAS_GE_210 else deprecate_ctx
+        ):
             result = getattr(ddf.groupby("group"), func)(axis=axis)
         assert_eq(expected, result)
 
@@ -2567,12 +2591,15 @@ def test_groupby_value_counts_10322():
 
 
 @contextlib.contextmanager
-def groupby_axis_and_meta():
+def groupby_axis_and_meta(axis=0):
     # Because we're checking for multiple warnings, we need to record
     # all warnings and inspect them after the fact
     with pytest.warns() as record:
         yield
-    assert len(record) == 2 if PANDAS_GE_210 else 1, [x.message for x in record.list]
+    expected_len = 2 if PANDAS_GE_210 else 1
+    if axis == 1:
+        expected_len += 1
+    assert expected_len, [x.message for x in record.list]
     assert record[-1].category is UserWarning
     assert "`meta` is not specified" in str(record[-1].message)
     if PANDAS_GE_210:
@@ -2595,22 +2622,28 @@ def test_groupby_shift_basic_input(npartitions, period, axis):
 
     with groupby_axis_deprecated():
         expected = pdf.groupby(["a", "c"]).shift(period, axis=axis)
-    with groupby_axis_and_meta():
+    with groupby_axis_and_meta(axis):
         result = ddf.groupby(["a", "c"]).shift(period, axis=axis)
     assert_eq(expected, result)
 
+    with pytest.warns(FutureWarning, match="`axis` parameter is deprecated"):
+        ddf.groupby(["a", "c"]).shift(period, axis=axis)
+
     with groupby_axis_deprecated():
         expected = pdf.groupby(["a"]).shift(period, axis=axis)
-    with groupby_axis_and_meta():
+    with groupby_axis_and_meta(axis):
         result = ddf.groupby(["a"]).shift(period, axis=axis)
     assert_eq(expected, result)
 
     with groupby_axis_deprecated():
         expected = pdf.groupby(pdf.c).shift(period, axis=axis)
-    with groupby_axis_and_meta():
+    with groupby_axis_and_meta(axis):
         result = ddf.groupby(ddf.c).shift(period, axis=axis)
     assert_eq(expected, result)
 
+    with pytest.warns(FutureWarning, match="`axis` parameter is deprecated"):
+        result = ddf.groupby(ddf.c).shift(period, axis=axis)
+
 
 def test_groupby_shift_series():
     pdf = pd.DataFrame(
diff --git a/dask/dataframe/tests/test_merge_column_and_index.py b/dask/dataframe/tests/test_merge_column_and_index.py
index bd455b61d..e8e1ac557 100644
--- a/dask/dataframe/tests/test_merge_column_and_index.py
+++ b/dask/dataframe/tests/test_merge_column_and_index.py
@@ -98,7 +98,7 @@ def test_merge_known_to_known(
     expected = df_left.merge(df_right, on=on, how=how)
 
     # Perform merge
-    result = ddf_left.merge(ddf_right, on=on, how=how, shuffle=shuffle_method)
+    result = ddf_left.merge(ddf_right, on=on, how=how, shuffle_method=shuffle_method)
 
     # Assertions
     assert_eq(result, expected)
@@ -114,7 +114,9 @@ def test_merge_known_to_single(
     expected = df_left.merge(df_right, on=on, how=how)
 
     # Perform merge
-    result = ddf_left.merge(ddf_right_single, on=on, how=how, shuffle=shuffle_method)
+    result = ddf_left.merge(
+        ddf_right_single, on=on, how=how, shuffle_method=shuffle_method
+    )
 
     # Assertions
     assert_eq(result, expected)
@@ -130,7 +132,9 @@ def test_merge_single_to_known(
     expected = df_left.merge(df_right, on=on, how=how)
 
     # Perform merge
-    result = ddf_left_single.merge(ddf_right, on=on, how=how, shuffle=shuffle_method)
+    result = ddf_left_single.merge(
+        ddf_right, on=on, how=how, shuffle_method=shuffle_method
+    )
 
     # Assertions
     assert_eq(result, expected)
@@ -145,7 +149,9 @@ def test_merge_known_to_unknown(
     expected = df_left.merge(df_right, on=on, how=how)
 
     # Perform merge
-    result = ddf_left.merge(ddf_right_unknown, on=on, how=how, shuffle=shuffle_method)
+    result = ddf_left.merge(
+        ddf_right_unknown, on=on, how=how, shuffle_method=shuffle_method
+    )
 
     # Assertions
     assert_eq(result, expected)
@@ -159,7 +165,9 @@ def test_merge_unknown_to_known(
     expected = df_left.merge(df_right, on=on, how=how)
 
     # Perform merge
-    result = ddf_left_unknown.merge(ddf_right, on=on, how=how, shuffle=shuffle_method)
+    result = ddf_left_unknown.merge(
+        ddf_right, on=on, how=how, shuffle_method=shuffle_method
+    )
 
     # Assertions
     assert_eq(result, expected)
@@ -180,7 +188,7 @@ def test_merge_unknown_to_unknown(
 
     # Merge unknown to unknown
     result = ddf_left_unknown.merge(
-        ddf_right_unknown, on=on, how=how, shuffle=shuffle_method
+        ddf_right_unknown, on=on, how=how, shuffle_method=shuffle_method
     )
 
     # Assertions
@@ -197,7 +205,7 @@ def test_merge_known_to_double_bcast_right(
 
     # Perform merge
     result = ddf_left.merge(
-        ddf_right_double, on=on, how=how, shuffle=shuffle_method, broadcast=True
+        ddf_right_double, on=on, how=how, shuffle_method=shuffle_method, broadcast=True
     )
 
     # Assertions
@@ -217,7 +225,7 @@ def test_merge_known_to_double_bcast_left(
 
     # Perform merge
     result = ddf_left_double.merge(
-        ddf_right, on=on, how=how, broadcast=broadcast, shuffle=shuffle_method
+        ddf_right, on=on, how=how, broadcast=broadcast, shuffle_method=shuffle_method
     )
 
     # Assertions
diff --git a/dask/dataframe/tests/test_multi.py b/dask/dataframe/tests/test_multi.py
index 38b3eed78..997fca113 100644
--- a/dask/dataframe/tests/test_multi.py
+++ b/dask/dataframe/tests/test_multi.py
@@ -258,7 +258,7 @@ def test_hash_join(how, shuffle_method):
     list_eq(result, expected)
 
     # Different columns and npartitions
-    c = hash_join(a, "x", b, "z", "outer", npartitions=3, shuffle=shuffle_method)
+    c = hash_join(a, "x", b, "z", "outer", npartitions=3, shuffle_method=shuffle_method)
     assert not hlg_layer_topological(c.dask, -1).is_materialized()
     assert c.npartitions == 3
 
@@ -268,12 +268,12 @@ def test_hash_join(how, shuffle_method):
     list_eq(result, expected)
 
     assert (
-        hash_join(a, "y", b, "y", "inner", shuffle=shuffle_method)._name
-        == hash_join(a, "y", b, "y", "inner", shuffle=shuffle_method)._name
+        hash_join(a, "y", b, "y", "inner", shuffle_method=shuffle_method)._name
+        == hash_join(a, "y", b, "y", "inner", shuffle_method=shuffle_method)._name
     )
     assert (
-        hash_join(a, "y", b, "y", "inner", shuffle=shuffle_method)._name
-        != hash_join(a, "y", b, "y", "outer", shuffle=shuffle_method)._name
+        hash_join(a, "y", b, "y", "inner", shuffle_method=shuffle_method)._name
+        != hash_join(a, "y", b, "y", "outer", shuffle_method=shuffle_method)._name
     )
 
 
@@ -952,7 +952,12 @@ def test_merge(how, shuffle_method):
 
     assert_eq(
         dd.merge(
-            a, b, left_index=True, right_index=True, how=how, shuffle=shuffle_method
+            a,
+            b,
+            left_index=True,
+            right_index=True,
+            how=how,
+            shuffle_method=shuffle_method,
         ),
         pd.merge(A, B, left_index=True, right_index=True, how=how),
     )
@@ -962,7 +967,9 @@ def test_merge(how, shuffle_method):
     assert all(d is None for d in result.divisions)
 
     list_eq(
-        dd.merge(a, b, left_on="x", right_on="z", how=how, shuffle=shuffle_method),
+        dd.merge(
+            a, b, left_on="x", right_on="z", how=how, shuffle_method=shuffle_method
+        ),
         pd.merge(A, B, left_on="x", right_on="z", how=how),
     )
     list_eq(
@@ -973,19 +980,32 @@ def test_merge(how, shuffle_method):
             right_on="z",
             how=how,
             suffixes=("1", "2"),
-            shuffle=shuffle_method,
+            shuffle_method=shuffle_method,
         ),
         pd.merge(A, B, left_on="x", right_on="z", how=how, suffixes=("1", "2")),
     )
 
-    list_eq(dd.merge(a, b, how=how, shuffle=shuffle_method), pd.merge(A, B, how=how))
-    list_eq(dd.merge(a, B, how=how, shuffle=shuffle_method), pd.merge(A, B, how=how))
-    list_eq(dd.merge(A, b, how=how, shuffle=shuffle_method), pd.merge(A, B, how=how))
-    list_eq(dd.merge(A, B, how=how, shuffle=shuffle_method), pd.merge(A, B, how=how))
+    list_eq(
+        dd.merge(a, b, how=how, shuffle_method=shuffle_method), pd.merge(A, B, how=how)
+    )
+    list_eq(
+        dd.merge(a, B, how=how, shuffle_method=shuffle_method), pd.merge(A, B, how=how)
+    )
+    list_eq(
+        dd.merge(A, b, how=how, shuffle_method=shuffle_method), pd.merge(A, B, how=how)
+    )
+    list_eq(
+        dd.merge(A, B, how=how, shuffle_method=shuffle_method), pd.merge(A, B, how=how)
+    )
 
     list_eq(
         dd.merge(
-            a, b, left_index=True, right_index=True, how=how, shuffle=shuffle_method
+            a,
+            b,
+            left_index=True,
+            right_index=True,
+            how=how,
+            shuffle_method=shuffle_method,
         ),
         pd.merge(A, B, left_index=True, right_index=True, how=how),
     )
@@ -997,13 +1017,15 @@ def test_merge(how, shuffle_method):
             right_index=True,
             how=how,
             suffixes=("1", "2"),
-            shuffle=shuffle_method,
+            shuffle_method=shuffle_method,
         ),
         pd.merge(A, B, left_index=True, right_index=True, how=how, suffixes=("1", "2")),
     )
 
     list_eq(
-        dd.merge(a, b, left_on="x", right_index=True, how=how, shuffle=shuffle_method),
+        dd.merge(
+            a, b, left_on="x", right_index=True, how=how, shuffle_method=shuffle_method
+        ),
         pd.merge(A, B, left_on="x", right_index=True, how=how),
     )
     list_eq(
@@ -1014,7 +1036,7 @@ def test_merge(how, shuffle_method):
             right_index=True,
             how=how,
             suffixes=("1", "2"),
-            shuffle=shuffle_method,
+            shuffle_method=shuffle_method,
         ),
         pd.merge(A, B, left_on="x", right_index=True, how=how, suffixes=("1", "2")),
     )
@@ -1024,6 +1046,26 @@ def test_merge(how, shuffle_method):
     #         pd.merge(A, B, left_index=True, right_on='y'))
 
 
+def test_merge_deprecated_shuffle_keyword(shuffle_method):
+    A = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6], "y": [1, 1, 2, 2, 3, 4]})
+    a = dd.repartition(A, [0, 4, 5])
+
+    B = pd.DataFrame({"y": [1, 3, 4, 4, 5, 6], "z": [6, 5, 4, 3, 2, 1]})
+    b = dd.repartition(B, [0, 2, 5])
+
+    expected = pd.merge(A, B, left_index=True, right_index=True)
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        result = dd.merge(
+            a, b, left_index=True, right_index=True, shuffle=shuffle_method
+        )
+    assert_eq(result, expected)
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        result = a.merge(b, left_index=True, right_index=True, shuffle=shuffle_method)
+    assert_eq(result, expected)
+
+
 @pytest.mark.parametrize("how", ["right", "outer"])
 def test_merge_empty_left_df(shuffle_method, how):
     # We're merging on "a", but in a situation where the left
@@ -1123,7 +1165,7 @@ def test_merge_tasks_passes_through():
     aa = dd.from_pandas(a, npartitions=3)
     bb = dd.from_pandas(b, npartitions=2)
 
-    cc = aa.merge(bb, left_on="a", right_on="d", shuffle="tasks")
+    cc = aa.merge(bb, left_on="a", right_on="d", shuffle_method="tasks")
 
     assert not any("partd" in k[0] for k in cc.dask)
 
@@ -1194,7 +1236,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pd.merge(pdl, pdr, how=how, left_index=True, right_index=True),
@@ -1208,7 +1250,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pd.merge(pdr, pdl, how=how, left_index=True, right_index=True),
@@ -1223,7 +1265,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                     indicator=True,
                 ),
                 fix_index(
@@ -1245,7 +1287,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                     indicator=True,
                 ),
                 fix_index(
@@ -1267,7 +1309,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pdr.merge(pdl, how=how, left_index=True, right_index=True),
@@ -1280,7 +1322,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pdl.merge(pdr, how=how, left_index=True, right_index=True),
@@ -1291,13 +1333,23 @@ def test_merge_by_index_patterns(how, shuffle_method):
             # hash join
             list_eq(
                 dd.merge(
-                    ddl, ddr, how=how, left_on="a", right_on="c", shuffle=shuffle_method
+                    ddl,
+                    ddr,
+                    how=how,
+                    left_on="a",
+                    right_on="c",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdl, pdr, how=how, left_on="a", right_on="c"),
             )
             list_eq(
                 dd.merge(
-                    ddl, ddr, how=how, left_on="b", right_on="d", shuffle=shuffle_method
+                    ddl,
+                    ddr,
+                    how=how,
+                    left_on="b",
+                    right_on="d",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdl, pdr, how=how, left_on="b", right_on="d"),
             )
@@ -1309,7 +1361,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_on="c",
                     right_on="a",
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                     indicator=True,
                 ),
                 pd.merge(pdr, pdl, how=how, left_on="c", right_on="a", indicator=True),
@@ -1321,7 +1373,7 @@ def test_merge_by_index_patterns(how, shuffle_method):
                     how=how,
                     left_on="d",
                     right_on="b",
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                     indicator=True,
                 ),
                 pd.merge(pdr, pdl, how=how, left_on="d", right_on="b", indicator=True),
@@ -1329,39 +1381,65 @@ def test_merge_by_index_patterns(how, shuffle_method):
 
             list_eq(
                 dd.merge(
-                    ddr, ddl, how=how, left_on="c", right_on="a", shuffle=shuffle_method
+                    ddr,
+                    ddl,
+                    how=how,
+                    left_on="c",
+                    right_on="a",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdr, pdl, how=how, left_on="c", right_on="a"),
             )
             list_eq(
                 dd.merge(
-                    ddr, ddl, how=how, left_on="d", right_on="b", shuffle=shuffle_method
+                    ddr,
+                    ddl,
+                    how=how,
+                    left_on="d",
+                    right_on="b",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdr, pdl, how=how, left_on="d", right_on="b"),
             )
 
             list_eq(
                 ddl.merge(
-                    ddr, how=how, left_on="a", right_on="c", shuffle=shuffle_method
+                    ddr,
+                    how=how,
+                    left_on="a",
+                    right_on="c",
+                    shuffle_method=shuffle_method,
                 ),
                 pdl.merge(pdr, how=how, left_on="a", right_on="c"),
             )
             list_eq(
                 ddl.merge(
-                    ddr, how=how, left_on="b", right_on="d", shuffle=shuffle_method
+                    ddr,
+                    how=how,
+                    left_on="b",
+                    right_on="d",
+                    shuffle_method=shuffle_method,
                 ),
                 pdl.merge(pdr, how=how, left_on="b", right_on="d"),
             )
 
             list_eq(
                 ddr.merge(
-                    ddl, how=how, left_on="c", right_on="a", shuffle=shuffle_method
+                    ddl,
+                    how=how,
+                    left_on="c",
+                    right_on="a",
+                    shuffle_method=shuffle_method,
                 ),
                 pdr.merge(pdl, how=how, left_on="c", right_on="a"),
             )
             list_eq(
                 ddr.merge(
-                    ddl, how=how, left_on="d", right_on="b", shuffle=shuffle_method
+                    ddl,
+                    how=how,
+                    left_on="d",
+                    right_on="b",
+                    shuffle_method=shuffle_method,
                 ),
                 pdr.merge(pdl, how=how, left_on="d", right_on="b"),
             )
@@ -1431,17 +1509,21 @@ def test_join_by_index_patterns(how, shuffle_method):
             ddr = dd.from_pandas(pdr, rpart)
 
             assert_eq(
-                ddl.join(ddr, how=how, shuffle=shuffle_method),
+                ddl.join(ddr, how=how, shuffle_method=shuffle_method),
                 fix_index(pdl.join(pdr, how=how), pdl.index.dtype),
             )
             assert_eq(
-                ddr.join(ddl, how=how, shuffle=shuffle_method),
+                ddr.join(ddl, how=how, shuffle_method=shuffle_method),
                 fix_index(pdr.join(pdl, how=how), pdr.index.dtype),
             )
 
             assert_eq(
                 ddl.join(
-                    ddr, how=how, lsuffix="l", rsuffix="r", shuffle=shuffle_method
+                    ddr,
+                    how=how,
+                    lsuffix="l",
+                    rsuffix="r",
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pdl.join(pdr, how=how, lsuffix="l", rsuffix="r"), pdl.index.dtype
@@ -1449,7 +1531,11 @@ def test_join_by_index_patterns(how, shuffle_method):
             )
             assert_eq(
                 ddr.join(
-                    ddl, how=how, lsuffix="l", rsuffix="r", shuffle=shuffle_method
+                    ddl,
+                    how=how,
+                    lsuffix="l",
+                    rsuffix="r",
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pdr.join(pdl, how=how, lsuffix="l", rsuffix="r"), pdl.index.dtype
@@ -1570,11 +1656,11 @@ def test_merge_by_multiple_columns(how, shuffle_method):
             ddr = dd.from_pandas(pdr, rpart)
 
             assert_eq(
-                ddl.join(ddr, how=how, shuffle=shuffle_method),
+                ddl.join(ddr, how=how, shuffle_method=shuffle_method),
                 fix_index(pdl.join(pdr, how=how), pdl.index.dtype),
             )
             assert_eq(
-                ddr.join(ddl, how=how, shuffle=shuffle_method),
+                ddr.join(ddl, how=how, shuffle_method=shuffle_method),
                 fix_index(pdr.join(pdl, how=how), pdr.index.dtype),
             )
 
@@ -1585,7 +1671,7 @@ def test_merge_by_multiple_columns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pd.merge(pdl, pdr, how=how, left_index=True, right_index=True),
@@ -1599,7 +1685,7 @@ def test_merge_by_multiple_columns(how, shuffle_method):
                     how=how,
                     left_index=True,
                     right_index=True,
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                 ),
                 fix_index(
                     pd.merge(pdr, pdl, how=how, left_index=True, right_index=True),
@@ -1610,26 +1696,46 @@ def test_merge_by_multiple_columns(how, shuffle_method):
             # hash join
             list_eq(
                 dd.merge(
-                    ddl, ddr, how=how, left_on="a", right_on="d", shuffle=shuffle_method
+                    ddl,
+                    ddr,
+                    how=how,
+                    left_on="a",
+                    right_on="d",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdl, pdr, how=how, left_on="a", right_on="d"),
             )
             list_eq(
                 dd.merge(
-                    ddl, ddr, how=how, left_on="b", right_on="e", shuffle=shuffle_method
+                    ddl,
+                    ddr,
+                    how=how,
+                    left_on="b",
+                    right_on="e",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdl, pdr, how=how, left_on="b", right_on="e"),
             )
 
             list_eq(
                 dd.merge(
-                    ddr, ddl, how=how, left_on="d", right_on="a", shuffle=shuffle_method
+                    ddr,
+                    ddl,
+                    how=how,
+                    left_on="d",
+                    right_on="a",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdr, pdl, how=how, left_on="d", right_on="a"),
             )
             list_eq(
                 dd.merge(
-                    ddr, ddl, how=how, left_on="e", right_on="b", shuffle=shuffle_method
+                    ddr,
+                    ddl,
+                    how=how,
+                    left_on="e",
+                    right_on="b",
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdr, pdl, how=how, left_on="e", right_on="b"),
             )
@@ -1641,7 +1747,7 @@ def test_merge_by_multiple_columns(how, shuffle_method):
                     how=how,
                     left_on=["a", "b"],
                     right_on=["d", "e"],
-                    shuffle=shuffle_method,
+                    shuffle_method=shuffle_method,
                 ),
                 pd.merge(pdl, pdr, how=how, left_on=["a", "b"], right_on=["d", "e"]),
             )
@@ -1814,7 +1920,7 @@ def test_merge_index_without_divisions(shuffle_method):
     aa = dd.from_pandas(a, npartitions=3, sort=False)
     bb = dd.from_pandas(b, npartitions=2)
 
-    result = aa.join(bb, how="inner", shuffle=shuffle_method)
+    result = aa.join(bb, how="inner", shuffle_method=shuffle_method)
     expected = a.join(b, how="inner")
     assert_eq(result, expected)
 
@@ -1829,7 +1935,7 @@ def test_half_indexed_dataframe_avoids_shuffle():
     bb = dd.from_pandas(b, npartitions=2)
 
     c = pd.merge(a, b, left_index=True, right_on="y")
-    cc = dd.merge(aa, bb, left_index=True, right_on="y", shuffle="tasks")
+    cc = dd.merge(aa, bb, left_index=True, right_on="y", shuffle_method="tasks")
 
     list_eq(c, cc)
 
@@ -2726,7 +2832,7 @@ def test_merge_tasks_large_to_small(how, npartitions, base):
         how=how,
         npartitions=npartitions,
         broadcast=broadcast_bias,
-        shuffle="tasks",
+        shuffle_method="tasks",
     )
     pd_result = pd.merge(left, right, on="y", how=how)
 
@@ -2751,7 +2857,7 @@ def test_broadcast_true(shuffle):
     left = dd.from_dict({"a": [1, 2] * 80, "b_left": range(160)}, npartitions=16)
     right = dd.from_dict({"a": [2, 1] * 10, "b_right": range(20)}, npartitions=2)
 
-    result = dd.merge(left, right, broadcast=True, shuffle=shuffle)
+    result = dd.merge(left, right, broadcast=True, shuffle_method=shuffle)
     assert hlg_layer(result.dask, "bcast-join")
 
 
diff --git a/dask/dataframe/tests/test_rolling.py b/dask/dataframe/tests/test_rolling.py
index a8fda109b..637e83458 100644
--- a/dask/dataframe/tests/test_rolling.py
+++ b/dask/dataframe/tests/test_rolling.py
@@ -361,11 +361,10 @@ def test_rolling_axis(kwargs):
     df = pd.DataFrame(np.random.randn(20, 16))
     ddf = dd.from_pandas(df, npartitions=3)
 
-    ctx = (
-        pytest.warns(FutureWarning, match="The 'axis' keyword|Support for axis")
-        if PANDAS_GE_210
-        else contextlib.nullcontext()
-    )
+    ctx = contextlib.nullcontext()
+    if PANDAS_GE_210:
+        ctx = pytest.warns(FutureWarning, match="The 'axis' keyword|Support for axis")
+
     if kwargs["axis"] == "series":
         # Series
         with ctx:
@@ -377,6 +376,8 @@ def test_rolling_axis(kwargs):
         # DataFrame
         with ctx:
             expected = df.rolling(3, **kwargs).mean()
+        if kwargs["axis"] in (1, "rows") and not PANDAS_GE_210:
+            ctx = pytest.warns(FutureWarning, match="Using axis=1 in Rolling")
         with ctx:
             result = ddf.rolling(3, **kwargs).mean()
         assert_eq(expected, result)
diff --git a/dask/dataframe/tests/test_shuffle.py b/dask/dataframe/tests/test_shuffle.py
index 05bec5513..cd3e173d1 100644
--- a/dask/dataframe/tests/test_shuffle.py
+++ b/dask/dataframe/tests/test_shuffle.py
@@ -61,7 +61,7 @@ shuffle_func = shuffle  # conflicts with keyword argument
 
 
 def test_shuffle(shuffle_method):
-    s = shuffle_func(d, d.b, shuffle=shuffle_method)
+    s = shuffle_func(d, d.b, shuffle_method=shuffle_method)
     assert isinstance(s, dd.DataFrame)
     assert s.npartitions == d.npartitions
 
@@ -74,6 +74,18 @@ def test_shuffle(shuffle_method):
     assert shuffle_func(d, d.b)._name == shuffle_func(d, d.b)._name
 
 
+def test_shuffle_deprecated_shuffle_keyword(shuffle_method):
+    from dask.dataframe.tests.test_multi import list_eq
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        result = dd.shuffle.shuffle(d, d.b, shuffle=shuffle_method)
+    list_eq(result, d)
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        result = d.shuffle(d.b, shuffle=shuffle_method)
+    list_eq(result, d)
+
+
 def test_default_partitions():
     assert shuffle(d, d.b).npartitions == d.npartitions
 
@@ -81,7 +93,7 @@ def test_default_partitions():
 def test_shuffle_npartitions(shuffle_method):
     df = pd.DataFrame({"x": np.random.random(100)})
     ddf = dd.from_pandas(df, npartitions=10)
-    s = shuffle(ddf, ddf.x, shuffle=shuffle_method, npartitions=17, max_branch=4)
+    s = shuffle(ddf, ddf.x, shuffle_method=shuffle_method, npartitions=17, max_branch=4)
     sc = s.compute()
     assert s.npartitions == 17
     assert set(s.dask).issuperset(set(ddf.dask))
@@ -94,7 +106,7 @@ def test_shuffle_npartitions(shuffle_method):
 def test_shuffle_npartitions_lt_input_partitions(shuffle_method):
     df = pd.DataFrame({"x": np.random.random(100)})
     ddf = dd.from_pandas(df, npartitions=20)
-    s = shuffle(ddf, ddf.x, shuffle=shuffle_method, npartitions=5, max_branch=2)
+    s = shuffle(ddf, ddf.x, shuffle_method=shuffle_method, npartitions=5, max_branch=2)
     sc = s.compute()
     assert s.npartitions == 5
     assert set(s.dask).issuperset(set(ddf.dask))
@@ -108,14 +120,15 @@ def test_index_with_non_series(shuffle_method):
     from dask.dataframe.tests.test_multi import list_eq
 
     list_eq(
-        shuffle(d, d.b, shuffle=shuffle_method), shuffle(d, "b", shuffle=shuffle_method)
+        shuffle(d, d.b, shuffle_method=shuffle_method),
+        shuffle(d, "b", shuffle_method=shuffle_method),
     )
 
 
 def test_index_with_dataframe(shuffle_method):
-    res1 = shuffle(d, d[["b"]], shuffle=shuffle_method).compute()
-    res2 = shuffle(d, ["b"], shuffle=shuffle_method).compute()
-    res3 = shuffle(d, "b", shuffle=shuffle_method).compute()
+    res1 = shuffle(d, d[["b"]], shuffle_method=shuffle_method).compute()
+    res2 = shuffle(d, ["b"], shuffle_method=shuffle_method).compute()
+    res3 = shuffle(d, "b", shuffle_method=shuffle_method).compute()
 
     assert sorted(res1.values.tolist()) == sorted(res2.values.tolist())
     assert sorted(res1.values.tolist()) == sorted(res3.values.tolist())
@@ -126,14 +139,14 @@ def test_shuffle_from_one_partition_to_one_other(shuffle_method):
     a = dd.from_pandas(df, 1)
 
     for i in [1, 2]:
-        b = shuffle(a, "x", npartitions=i, shuffle=shuffle_method)
+        b = shuffle(a, "x", npartitions=i, shuffle_method=shuffle_method)
         assert len(a.compute(scheduler="sync")) == len(b.compute(scheduler="sync"))
 
 
 def test_shuffle_empty_partitions(shuffle_method):
     df = pd.DataFrame({"x": [1, 2, 3] * 10})
     ddf = dd.from_pandas(df, npartitions=3)
-    s = shuffle(ddf, ddf.x, npartitions=6, shuffle=shuffle_method)
+    s = shuffle(ddf, ddf.x, npartitions=6, shuffle_method=shuffle_method)
     parts = compute_as_if_collection(dd.DataFrame, s.dask, s.__dask_keys__())
     for p in parts:
         assert s.columns == p.columns
@@ -208,14 +221,20 @@ def test_set_index_general(npartitions, shuffle_method):
 
     ddf = dd.from_pandas(df, npartitions=npartitions)
 
-    assert_eq(df.set_index("x"), ddf.set_index("x", shuffle=shuffle_method))
-    assert_eq(df.set_index("y"), ddf.set_index("y", shuffle=shuffle_method))
-    assert_eq(df.set_index("z"), ddf.set_index("z", shuffle=shuffle_method))
-    assert_eq(df.set_index(df.x), ddf.set_index(ddf.x, shuffle=shuffle_method))
+    assert_eq(df.set_index("x"), ddf.set_index("x", shuffle_method=shuffle_method))
+    assert_eq(df.set_index("y"), ddf.set_index("y", shuffle_method=shuffle_method))
+    assert_eq(df.set_index("z"), ddf.set_index("z", shuffle_method=shuffle_method))
+    assert_eq(df.set_index(df.x), ddf.set_index(ddf.x, shuffle_method=shuffle_method))
     assert_eq(
-        df.set_index(df.x + df.y), ddf.set_index(ddf.x + ddf.y, shuffle=shuffle_method)
+        df.set_index(df.x + df.y),
+        ddf.set_index(ddf.x + ddf.y, shuffle_method=shuffle_method),
+    )
+    assert_eq(
+        df.set_index(df.x + 1), ddf.set_index(ddf.x + 1, shuffle_method=shuffle_method)
     )
-    assert_eq(df.set_index(df.x + 1), ddf.set_index(ddf.x + 1, shuffle=shuffle_method))
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        assert_eq(df.set_index("x"), ddf.set_index("x", shuffle=shuffle_method))
 
 
 @pytest.mark.skipif(
@@ -237,7 +256,7 @@ def test_set_index_string(shuffle_method, string_dtype):
     )
     df = df.astype({"y": string_dtype})
     ddf = dd.from_pandas(df, npartitions=10)
-    assert_eq(df.set_index("y"), ddf.set_index("y", shuffle=shuffle_method))
+    assert_eq(df.set_index("y"), ddf.set_index("y", shuffle_method=shuffle_method))
 
 
 def test_set_index_self_index(shuffle_method):
@@ -248,7 +267,7 @@ def test_set_index_self_index(shuffle_method):
 
     a = dd.from_pandas(df, npartitions=4)
     with pytest.warns(UserWarning, match="this is a no-op"):
-        b = a.set_index(a.index, shuffle=shuffle_method)
+        b = a.set_index(a.index, shuffle_method=shuffle_method)
     assert a is b
 
     assert_eq(b, df.set_index(df.index))
@@ -265,18 +284,18 @@ def test_set_index_names(shuffle_method):
 
     ddf = dd.from_pandas(df, npartitions=4)
 
-    assert set(ddf.set_index("x", shuffle=shuffle_method).dask) == set(
-        ddf.set_index("x", shuffle=shuffle_method).dask
-    )
-    assert set(ddf.set_index("x", shuffle=shuffle_method).dask) != set(
-        ddf.set_index("y", shuffle=shuffle_method).dask
+    assert set(ddf.set_index("x", shuffle_method=shuffle_method).dask) == set(
+        ddf.set_index("x", shuffle_method=shuffle_method).dask
     )
-    assert set(ddf.set_index("x", max_branch=4, shuffle=shuffle_method).dask) != set(
-        ddf.set_index("x", max_branch=3, shuffle=shuffle_method).dask
-    )
-    assert set(ddf.set_index("x", drop=True, shuffle=shuffle_method).dask) != set(
-        ddf.set_index("x", drop=False, shuffle=shuffle_method).dask
+    assert set(ddf.set_index("x", shuffle_method=shuffle_method).dask) != set(
+        ddf.set_index("y", shuffle_method=shuffle_method).dask
     )
+    assert set(
+        ddf.set_index("x", max_branch=4, shuffle_method=shuffle_method).dask
+    ) != set(ddf.set_index("x", max_branch=3, shuffle_method=shuffle_method).dask)
+    assert set(
+        ddf.set_index("x", drop=True, shuffle_method=shuffle_method).dask
+    ) != set(ddf.set_index("x", drop=False, shuffle_method=shuffle_method).dask)
 
 
 ME = "ME" if PANDAS_GE_220 else "M"
@@ -292,7 +311,7 @@ def test_set_index_2(shuffle_method):
         seed=1,
     )
 
-    df2 = df.set_index("name", shuffle=shuffle_method)
+    df2 = df.set_index("name", shuffle_method=shuffle_method)
     df2.value.sum().compute(scheduler="sync")
 
 
@@ -301,7 +320,7 @@ def test_set_index_3(shuffle_method):
     ddf = dd.from_pandas(df, npartitions=5)
 
     ddf2 = ddf.set_index(
-        "x", shuffle=shuffle_method, max_branch=2, npartitions=ddf.npartitions
+        "x", shuffle_method=shuffle_method, max_branch=2, npartitions=ddf.npartitions
     )
     df2 = df.set_index("x")
     assert_eq(df2, ddf2)
@@ -341,7 +360,7 @@ def test_shuffle_sort(shuffle_method):
     ddf = dd.from_pandas(df, npartitions=3)
 
     df2 = df.set_index("x").sort_index()
-    ddf2 = ddf.set_index("x", shuffle=shuffle_method)
+    ddf2 = ddf.set_index("x", shuffle_method=shuffle_method)
 
     assert_eq(ddf2.loc[2:3], df2.loc[2:3])
 
@@ -353,7 +372,7 @@ def test_rearrange(shuffle_method, scheduler):
     ddf2 = ddf.assign(_partitions=ddf.x % 4)
 
     result = rearrange_by_column(
-        ddf2, "_partitions", max_branch=32, shuffle=shuffle_method
+        ddf2, "_partitions", max_branch=32, shuffle_method=shuffle_method
     )
     assert result.npartitions == ddf.npartitions
     assert set(ddf.dask).issubset(result.dask)
@@ -375,7 +394,9 @@ def test_rearrange_cleanup():
     tmpdir = tempfile.mkdtemp()
 
     with dask.config.set(temporay_directory=str(tmpdir)):
-        result = rearrange_by_column(ddf2, "_partitions", max_branch=32, shuffle="disk")
+        result = rearrange_by_column(
+            ddf2, "_partitions", max_branch=32, shuffle_method="disk"
+        )
         result.compute(scheduler="processes")
 
     assert len(os.listdir(tmpdir)) == 0
@@ -398,7 +419,7 @@ def test_rearrange_disk_cleanup_with_exception():
         with dask.config.set(temporay_directory=str(tmpdir)):
             with pytest.raises(ValueError, match="Mock exception!"):
                 result = rearrange_by_column(
-                    ddf2, "_partitions", max_branch=32, shuffle="disk"
+                    ddf2, "_partitions", max_branch=32, shuffle_method="disk"
                 )
                 result.compute(scheduler="processes")
 
@@ -552,7 +573,7 @@ def test_set_index_reduces_partitions_small(shuffle_method):
     df = pd.DataFrame({"x": np.random.random(100)})
     ddf = dd.from_pandas(df, npartitions=50)
 
-    ddf2 = ddf.set_index("x", shuffle=shuffle_method, npartitions="auto")
+    ddf2 = ddf.set_index("x", shuffle_method=shuffle_method, npartitions="auto")
     assert ddf2.npartitions < 10
 
 
@@ -571,7 +592,7 @@ def test_set_index_reduces_partitions_large(shuffle_method):
         [None] * (nparts + 1),
     )
     ddf2 = ddf.set_index(
-        "x", shuffle=shuffle_method, npartitions="auto", partition_size=nbytes
+        "x", shuffle_method=shuffle_method, npartitions="auto", partition_size=nbytes
     )
     assert 1 < ddf2.npartitions < 20
 
@@ -587,7 +608,7 @@ def test_set_index_doesnt_increase_partitions(shuffle_method):
         [None] * (nparts + 1),
     )
     ddf2 = ddf.set_index(
-        "x", shuffle=shuffle_method, npartitions="auto", partition_size=nbytes
+        "x", shuffle_method=shuffle_method, npartitions="auto", partition_size=nbytes
     )
     assert ddf2.npartitions <= ddf.npartitions
 
@@ -596,7 +617,7 @@ def test_set_index_detects_sorted_data(shuffle_method):
     df = pd.DataFrame({"x": range(100), "y": range(100)})
     ddf = dd.from_pandas(df, npartitions=10, name="x", sort=False)
 
-    ddf2 = ddf.set_index("x", shuffle=shuffle_method)
+    ddf2 = ddf.set_index("x", shuffle_method=shuffle_method)
     assert len(ddf2.dask) < ddf.npartitions * 4
 
 
@@ -719,6 +740,21 @@ def test_set_index(engine):
     assert_eq(d5, full.set_index(["b"]))
 
 
+def test_set_index_deprecated_shuffle_keyword(shuffle_method):
+    df = pd.DataFrame({"x": [4, 1, 1, 3, 3], "y": [1.0, 1, 1, 1, 2]})
+    ddf = dd.from_pandas(df, 2)
+
+    expected = df.set_index("x")
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        result = ddf.set_index("x", shuffle=shuffle_method)
+    assert_eq(result, expected)
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        result = dd.shuffle.set_index(ddf, "x", shuffle=shuffle_method)
+    assert_eq(result, expected)
+
+
 @pytest.mark.parametrize(
     "engine", ["pandas", pytest.param("cudf", marks=pytest.mark.gpu)]
 )
@@ -1201,9 +1237,14 @@ def test_dataframe_shuffle_on_arg(on, ignore_index, max_branch, shuffle_method):
     else:
         ext_on = df_in[on].copy()
     df_out_1 = df_in.shuffle(
-        on, shuffle=shuffle_method, ignore_index=ignore_index, max_branch=max_branch
+        on,
+        shuffle_method=shuffle_method,
+        ignore_index=ignore_index,
+        max_branch=max_branch,
+    )
+    df_out_2 = df_in.shuffle(
+        ext_on, shuffle_method=shuffle_method, ignore_index=ignore_index
     )
-    df_out_2 = df_in.shuffle(ext_on, shuffle=shuffle_method, ignore_index=ignore_index)
 
     assert_eq(df_out_1, df_out_2, check_index=(not ignore_index))
 
@@ -1295,7 +1336,7 @@ def test_shuffle_hlg_layer():
         pd.DataFrame({"a": np.random.randint(0, 10, 100)}), npartitions=10
     )
     # Disk-based shuffle doesn't use HLG layers at the moment, so we only test tasks
-    ddf_shuffled = ddf.shuffle("a", max_branch=3, shuffle="tasks")
+    ddf_shuffled = ddf.shuffle("a", max_branch=3, shuffle_method="tasks")
     keys = [(ddf_shuffled._name, i) for i in range(ddf_shuffled.npartitions)]
 
     # Cull the HLG
@@ -1334,7 +1375,7 @@ def test_shuffle_partitions_meta_dtype():
         npartitions=10,
     )
     # Disk-based shuffle doesn't use HLG layers at the moment, so we only test tasks
-    ddf_shuffled = ddf.shuffle(ddf["a"] % 10, max_branch=3, shuffle="tasks")
+    ddf_shuffled = ddf.shuffle(ddf["a"] % 10, max_branch=3, shuffle_method="tasks")
     # Cull the HLG
     dsk = ddf_shuffled.__dask_graph__()
 
@@ -1355,7 +1396,7 @@ def test_shuffle_hlg_layer_serialize(npartitions):
         pd.DataFrame({"a": np.random.randint(0, 10, 100)}), npartitions=npartitions
     )
     # Disk-based shuffle doesn't use HLG layers at the moment, so we only test tasks
-    ddf_shuffled = ddf.shuffle("a", max_branch=3, shuffle="tasks")
+    ddf_shuffled = ddf.shuffle("a", max_branch=3, shuffle_method="tasks")
 
     # Ensure shuffle layers can be serialized and don't result in
     # the underlying low-level graph being materialized
@@ -1440,7 +1481,7 @@ def test_set_index_partitions_meta_dtype():
         npartitions=10,
     )
     # Disk-based shuffle doesn't use HLG layers at the moment, so we only test tasks
-    ddf = ddf.set_index("a", shuffle="tasks")
+    ddf = ddf.set_index("a", shuffle_method="tasks")
     # Cull the HLG
     dsk = ddf.__dask_graph__()
 
@@ -1462,7 +1503,7 @@ def test_sort_values_partitions_meta_dtype_with_divisions():
             npartitions=10,
         )
         # Disk-based shuffle doesn't use HLG layers at the moment, so we only test tasks
-        ddf = ddf.set_index("a", shuffle="tasks").sort_values("b")
+        ddf = ddf.set_index("a", shuffle_method="tasks").sort_values("b")
         # Cull the HLG
         dsk = ddf.__dask_graph__()
 
@@ -1488,6 +1529,19 @@ def test_sort_values(nelem, by, ascending):
     dd.assert_eq(got, expect, check_index=False, sort_results=False)
 
 
+def test_sort_values_deprecated_shuffle_keyword(shuffle_method):
+    np.random.seed(0)
+    df = pd.DataFrame()
+    df["a"] = np.ascontiguousarray(np.arange(10)[::-1])
+    df["b"] = np.arange(100, 10 + 100)
+    ddf = dd.from_pandas(df, npartitions=10)
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        got = ddf.sort_values(by=["a"], shuffle=shuffle_method)
+    expect = df.sort_values(by=["a"])
+    dd.assert_eq(got, expect, check_index=False, sort_results=False)
+
+
 @pytest.mark.parametrize(
     "backend", ["pandas", pytest.param("cudf", marks=pytest.mark.gpu)]
 )
@@ -1502,7 +1556,17 @@ def test_sort_values_tasks_backend(backend, by, ascending):
     ddf = dd.from_pandas(pdf, npartitions=10).to_backend(backend)
 
     expect = pdf.sort_values(by=by, ascending=ascending)
-    got = dd.DataFrame.sort_values(ddf, by=by, ascending=ascending, shuffle="tasks")
+    got = dd.DataFrame.sort_values(
+        ddf, by=by, ascending=ascending, shuffle_method="tasks"
+    )
+    dd.assert_eq(got, expect, sort_results=False)
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        got = dd.DataFrame.sort_values(ddf, by=by, ascending=ascending, shuffle="tasks")
+    dd.assert_eq(got, expect, sort_results=False)
+
+    with pytest.warns(FutureWarning, match="'shuffle' keyword is deprecated"):
+        got = ddf.sort_values(by=by, ascending=ascending, shuffle="tasks")
     dd.assert_eq(got, expect, sort_results=False)
 
 
diff --git a/dask/tests/test_distributed.py b/dask/tests/test_distributed.py
index d8b3c0b6b..ba5704368 100644
--- a/dask/tests/test_distributed.py
+++ b/dask/tests/test_distributed.py
@@ -147,7 +147,7 @@ def test_fused_blockwise_dataframe_merge(c, fuse):
     df2 += 10
 
     with dask.config.set({"optimization.fuse.active": fuse}):
-        ddfm = ddf1.merge(ddf2, on=["x"], how="left", shuffle="tasks")
+        ddfm = ddf1.merge(ddf2, on=["x"], how="left", shuffle_method="tasks")
         ddfm.head()  # https://github.com/dask/dask/issues/7178
         dfm = ddfm.compute().sort_values("x")
         # We call compute above since `sort_values` is not
@@ -169,7 +169,7 @@ def test_dataframe_broadcast_merge(c, on, broadcast):
     dfl = dd.from_pandas(pdfl, npartitions=4)
     dfr = dd.from_pandas(pdfr, npartitions=2)
 
-    ddfm = dd.merge(dfl, dfr, on=on, broadcast=broadcast, shuffle="tasks")
+    ddfm = dd.merge(dfl, dfr, on=on, broadcast=broadcast, shuffle_method="tasks")
     dfm = ddfm.compute()
     dd.utils.assert_eq(
         dfm.sort_values("a"),
@@ -613,7 +613,7 @@ async def test_combo_of_layer_types(c, s, a, b):
     )
 
     df = dd.from_pandas(pd.DataFrame({"a": np.arange(3)}), npartitions=3)
-    df = df.shuffle("a", shuffle="tasks")
+    df = df.shuffle("a", shuffle_method="tasks")
     df = df["a"].to_dask_array()
 
     res = x.sum() + df.sum()
@@ -719,7 +719,7 @@ async def test_shuffle_priority(c, s, a, b, max_branch, expected_layer_type):
     df = pd.DataFrame({"a": range(1000)})
     ddf = dd.from_pandas(df, npartitions=10)
 
-    ddf2 = ddf.shuffle("a", shuffle="tasks", max_branch=max_branch)
+    ddf2 = ddf.shuffle("a", shuffle_method="tasks", max_branch=max_branch)
 
     shuffle_layers = set(ddf2.dask.layers) - set(ddf.dask.layers)
     for layer_name in shuffle_layers:
@@ -765,7 +765,7 @@ def test_map_partitions_df_input():
         merged_df = dd.from_pandas(pd.DataFrame({"b": range(10)}), npartitions=1)
 
         # Notice, we include a shuffle in order to trigger a complex culling
-        merged_df = merged_df.shuffle(on="b", shuffle="tasks")
+        merged_df = merged_df.shuffle(on="b", shuffle_method="tasks")
 
         merged_df.map_partitions(
             f, ddf, meta=merged_df, enforce_metadata=False
diff --git a/dask/tests/test_layers.py b/dask/tests/test_layers.py
index e55e33e11..cfc9f88f7 100644
--- a/dask/tests/test_layers.py
+++ b/dask/tests/test_layers.py
@@ -78,7 +78,7 @@ def _dataframe_shuffle(tmpdir):
 
     # Perform a computation using an HLG-based shuffle
     df = pd.DataFrame({"a": range(10), "b": range(10, 20)})
-    return dd.from_pandas(df, npartitions=2).shuffle("a", shuffle="tasks")
+    return dd.from_pandas(df, npartitions=2).shuffle("a", shuffle_method="tasks")
 
 
 def _dataframe_tree_reduction(tmpdir):
@@ -98,7 +98,7 @@ def _dataframe_broadcast_join(tmpdir):
     df = pd.DataFrame({"a": range(10), "b": range(10, 20)})
     ddf1 = dd.from_pandas(df, npartitions=4)
     ddf2 = dd.from_pandas(df, npartitions=1)
-    return ddf1.merge(ddf2, how="left", broadcast=True, shuffle="tasks")
+    return ddf1.merge(ddf2, how="left", broadcast=True, shuffle_method="tasks")
 
 
 def _array_creation(tmpdir):
@@ -232,7 +232,7 @@ def test_scheduler_highlevel_graph_unpack_import(op, lib, optimize_graph, loop,
 
 
 def _shuffle_op(ddf):
-    return ddf.shuffle("x", shuffle="tasks")
+    return ddf.shuffle("x", shuffle_method="tasks")
 
 
 def _groupby_op(ddf):