Repository: dask/dask

Issue:
Deprecate fastparquet engine for read_parquet to enable switch to dask-expr
dask-expr will only support ``engine="pyarrow"``, so we should deprecate this here to allow users time to switch the engine before we switch to dask-expr or at least know that this change is coming.

Discussion:

Patch:
diff --git a/continuous_integration/environment-3.10.yaml b/continuous_integration/environment-3.10.yaml
index 25a7bf2b1..7b8ee6ef0 100644
--- a/continuous_integration/environment-3.10.yaml
+++ b/continuous_integration/environment-3.10.yaml
@@ -27,7 +27,7 @@ dependencies:
   - numpy=1.23
   - pandas=1.5
   - flask
-  - fastparquet>=0.8.0
+  - fastparquet
   - h5py
   - pytables
   - zarr
diff --git a/continuous_integration/environment-3.11.yaml b/continuous_integration/environment-3.11.yaml
index 8157f839f..56e81a405 100644
--- a/continuous_integration/environment-3.11.yaml
+++ b/continuous_integration/environment-3.11.yaml
@@ -27,7 +27,7 @@ dependencies:
   - numpy
   - pandas
   - flask
-  - fastparquet>=0.8.0
+  - fastparquet
   - h5py
   - pytables
   - zarr
diff --git a/continuous_integration/environment-3.12.yaml b/continuous_integration/environment-3.12.yaml
index 85b32b2b7..cab691854 100644
--- a/continuous_integration/environment-3.12.yaml
+++ b/continuous_integration/environment-3.12.yaml
@@ -27,7 +27,7 @@ dependencies:
   - numpy
   - pandas
   - flask
-  - fastparquet>=0.8.0
+  - fastparquet
   - h5py
   - pytables
   - zarr
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 03d451225..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,
         )
@@ -5338,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.
@@ -5809,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,
@@ -5821,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
@@ -5875,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.
@@ -5931,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,
@@ -5944,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()
@@ -5969,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
@@ -5980,7 +5984,7 @@ class DataFrame(_Frame):
                     lsuffix=lsuffix,
                     rsuffix=rsuffix,
                     npartitions=npartitions,
-                    shuffle=shuffle,
+                    shuffle_method=shuffle_method,
                 )
 
         from dask.dataframe.multi import merge
@@ -5994,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 ba1444326..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,
             )
 
@@ -2028,7 +2028,7 @@ class _GroupBy:
             },
             split_every=split_every,
             split_out=split_out,
-            shuffle=shuffle,
+            shuffle_method=shuffle,
             sort=self.sort,
         )
 
@@ -2358,7 +2358,7 @@ class _GroupBy:
                     token="aggregate",
                     split_every=split_every,
                     split_out=split_out,
-                    shuffle=shuffle,
+                    shuffle_method=shuffle,
                     sort=self.sort,
                 )
             else:
@@ -2382,7 +2382,7 @@ class _GroupBy:
                     token="aggregate",
                     split_every=split_every,
                     split_out=split_out,
-                    shuffle=shuffle,
+                    shuffle_method=shuffle,
                     sort=self.sort,
                 )
         else:
@@ -3193,7 +3193,7 @@ def _shuffle_aggregate(
     split_out=1,
     sort=True,
     ignore_index=False,
-    shuffle="tasks",
+    shuffle_method="tasks",
 ):
     """Shuffle-based groupby aggregation
 
@@ -3229,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:
@@ -3309,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,
@@ -3320,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/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index 209ad3fa4..37319708e 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -1706,7 +1706,7 @@ class ArrowDatasetEngine(Engine):
             frag = None
 
             # Check if we have partitioning information.
-            # Will only have this if the engine="pyarrow-dataset"
+            # Not supported with fastparquet.
             partitioning = kwargs.get("dataset", {}).get("partitioning", None)
 
             # Check if we need to generate a fragment.
diff --git a/dask/dataframe/io/parquet/core.py b/dask/dataframe/io/parquet/core.py
index d26a266d6..ad15de736 100644
--- a/dask/dataframe/io/parquet/core.py
+++ b/dask/dataframe/io/parquet/core.py
@@ -3,6 +3,7 @@ from __future__ import annotations
 import contextlib
 import math
 import warnings
+from typing import Literal
 
 import pandas as pd
 import tlz as toolz
@@ -220,23 +221,16 @@ def read_parquet(
         read in the data as a Series.
     filters : Union[List[Tuple[str, str, Any]], List[List[Tuple[str, str, Any]]]], default None
         List of filters to apply, like ``[[('col1', '==', 0), ...], ...]``.
-        Using this argument will NOT result in row-wise filtering of the final
-        partitions unless ``engine="pyarrow"`` is also specified.  For
-        other engines, filtering is only performed at the partition level, that is,
-        to prevent the loading of some row-groups and/or files.
+        Using this argument will result in row-wise filtering of the final partitions.
 
-        For the "pyarrow" engine, predicates can be expressed in disjunctive
-        normal form (DNF). This means that the inner-most tuple describes a single
-        column predicate. These inner predicates are combined with an AND
-        conjunction into a larger predicate. The outer-most list then combines all
-        of the combined filters with an OR disjunction.
+        Predicates can be expressed in disjunctive normal form (DNF). This means that
+        the inner-most tuple describes a single column predicate. These inner predicates
+        are combined with an AND conjunction into a larger predicate. The outer-most
+        list then combines all of the combined filters with an OR disjunction.
 
         Predicates can also be expressed as a ``List[Tuple]``. These are evaluated
         as an AND conjunction. To express OR in predicates, one must use the
         (preferred for "pyarrow") ``List[List[Tuple]]`` notation.
-
-        Note that the "fastparquet" engine does not currently support DNF for
-        the filtering of partitioned columns (``List[Tuple]`` is required).
     index : str, list or False, default None
         Field name(s) to use as the output frame index. By default will be
         inferred from the pandas parquet file metadata, if present. Use ``False``
@@ -247,7 +241,7 @@ def read_parquet(
         guaranteed that the column is encoded as dictionary in all row-groups.
         If a list, assumes up to 2**16-1 labels; if a dict, specify the number
         of labels expected; if None, will load categories automatically for
-        data written by dask/fastparquet, not otherwise.
+        data written by dask, not otherwise.
     storage_options : dict, default None
         Key/value pairs to be passed on to the file-system backend, if any.
         Note that the default file-system backend can be configured with the
@@ -260,12 +254,12 @@ def read_parquet(
         ``"precache_options"`` key. Also, a custom file-open function can be
         used (instead of ``AbstractFileSystem.open``), by specifying the
         desired function under the ``"open_file_func"`` key.
-    engine : {'auto', 'pyarrow', 'fastparquet'}, default 'auto'
+    engine : {'auto', 'pyarrow'}
         Parquet library to use. Defaults to 'auto', which uses ``pyarrow`` if
-        it is installed, and falls back to ``fastparquet`` otherwise.
+        it is installed, and falls back to the deprecated ``fastparquet`` otherwise.
+        This is also used by third-party packages (e.g. CuDF) to inject bespoke engines.
     use_nullable_dtypes : {False, True}
         Whether to use extension dtypes for the resulting ``DataFrame``.
-        ``use_nullable_dtypes=True`` is only supported when ``engine="pyarrow"``.
 
         .. note::
 
@@ -314,8 +308,7 @@ def read_parquet(
         (uncompressed) parquet storage space. This argument is currenlty used to
         set the default value of ``split_row_groups`` (using row-group metadata
         from a single file), and will be ignored if ``split_row_groups`` is not
-        set to 'infer' or 'adaptive'. Default may be engine-dependant, but is
-        256 MiB for the 'pyarrow' and 'fastparquet' engines.
+        set to 'infer' or 'adaptive'. Default is 256 MiB.
     aggregate_files : bool or str, default None
         WARNING: Passing a string argument to ``aggregate_files`` will result
         in experimental behavior. This behavior may change in the future.
@@ -361,17 +354,14 @@ def read_parquet(
         It may be necessary to change this argument if the data files in your
         parquet dataset do not end in ".parq", ".parquet", or ".pq".
     filesystem: "fsspec", "arrow", or fsspec.AbstractFileSystem backend to use.
-        Note that the "fastparquet" engine only supports "fsspec" or an explicit
-        ``pyarrow.fs.AbstractFileSystem`` object. Default is "fsspec".
     dataset: dict, default None
-        Dictionary of options to use when creating a ``pyarrow.dataset.Dataset``
-        or ``fastparquet.ParquetFile`` object. These options may include a
-        "filesystem" key (or "fs" for the "fastparquet" engine) to configure
-        the desired file-system backend. However, the top-level ``filesystem``
-        argument will always take precedence.
-
-        NOTE: For the "pyarrow" engine, the ``dataset`` options may include a
-        "partitioning" key. However, since ``pyarrow.dataset.Partitioning``
+        Dictionary of options to use when creating a ``pyarrow.dataset.Dataset`` object.
+        These options may include a "filesystem" key to configure the desired
+        file-system backend. However, the top-level ``filesystem`` argument will always
+        take precedence.
+
+        **Note**: The ``dataset`` options may include a "partitioning" key.
+        However, since ``pyarrow.dataset.Partitioning``
         objects cannot be serialized, the value can be a dict of key-word
         arguments for the ``pyarrow.dataset.partitioning`` API
         (e.g. ``dataset={"partitioning": {"flavor": "hive", "schema": ...}}``).
@@ -493,8 +483,7 @@ def read_parquet(
     if columns is not None:
         columns = list(columns)
 
-    if isinstance(engine, str):
-        engine = get_engine(engine)
+    engine = get_engine(engine)
 
     if hasattr(path, "name"):
         path = stringify_path(path)
@@ -726,9 +715,10 @@ def to_parquet(
     path : string or pathlib.Path
         Destination directory for data.  Prepend with protocol like ``s3://``
         or ``hdfs://`` for remote data.
-    engine : {'auto', 'pyarrow', 'fastparquet'}, default 'auto'
+    engine : {'auto', 'pyarrow'}
         Parquet library to use. Defaults to 'auto', which uses ``pyarrow`` if
-        it is installed, and falls back to ``fastparquet`` otherwise.
+        it is installed, and falls back to the deprecated ``fastparquet`` otherwise.
+        This is also used by third-party packages (e.g. CuDF) to inject bespoke engines.
     compression : string or dict, default 'snappy'
         Either a string like ``"snappy"`` or a dictionary mapping column names
         to compressors like ``{"name": "gzip", "values": "snappy"}``. Defaults
@@ -782,8 +772,7 @@ def to_parquet(
         no schema inference will be done. Passing in ``schema=None`` will
         disable the use of a global file schema - each written file may use a
         different schema dependent on the dtypes of the corresponding
-        partition. Note that this argument is ignored by the "fastparquet"
-        engine.
+        partition.
     name_function : callable, default None
         Function to generate the filename for each output partition.
         The function should accept an integer (partition index) as input and
@@ -793,8 +782,6 @@ def to_parquet(
         ``part.0.parquet``, ``part.1.parquet``, ``part.2.parquet``, ...
         and so on for each partition in the DataFrame.
     filesystem: "fsspec", "arrow", or fsspec.AbstractFileSystem backend to use.
-        Note that the "fastparquet" engine only supports "fsspec" or an explicit
-        ``pyarrow.fs.AbstractFileSystem`` object. Default is "fsspec".
     **kwargs :
         Extra options to be passed on to the specific backend.
 
@@ -845,8 +832,7 @@ def to_parquet(
     if df.columns.inferred_type not in {"string", "empty"}:
         raise ValueError("parquet doesn't support non-string column names")
 
-    if isinstance(engine, str):
-        engine = get_engine(engine)
+    engine = get_engine(engine)
 
     if hasattr(path, "name"):
         path = stringify_path(path)
@@ -1084,8 +1070,6 @@ def create_metadata_file(
     utility provides a mechanism to generate a _metadata file from a
     list of existing parquet files.
 
-    NOTE: This utility is not yet supported for the "fastparquet" engine.
-
     Parameters
     ----------
     paths : list(string)
@@ -1124,14 +1108,14 @@ def create_metadata_file(
     """
 
     # Get engine.
-    # Note that "fastparquet" is not yet supported
+    # Note that "fastparquet" is not supported.
     if isinstance(engine, str):
-        if engine not in ("pyarrow", "arrow"):
+        engine = get_engine(engine)
+        if engine is not _ENGINES.get("pyarrow"):
             raise ValueError(
-                f"{engine} is not a supported engine for create_metadata_file "
-                "Try engine='pyarrow'."
+                "fastparquet is not a supported engine for create_metadata_file."
+                "Please install pyarrow."
             )
-        engine = get_engine(engine)
 
     # Process input path list
     if fs is None:
@@ -1206,60 +1190,83 @@ def create_metadata_file(
     return out
 
 
-_ENGINES: dict[str, Engine] = {}
+_ENGINES: dict[str, type[Engine]] = {}
 
 
-def get_engine(engine):
+def get_engine(engine: Literal["auto", "pyarrow"] | type[Engine]) -> type[Engine]:
     """Get the parquet engine backend implementation.
 
     Parameters
     ----------
-    engine : {'auto', 'pyarrow', 'fastparquet'}, default 'auto'
+    engine : {'auto', 'pyarrow', 'fastparquet'} or Engine subclass
         Parquet library to use. Defaults to 'auto', which uses ``pyarrow`` if
-        it is installed, and falls back to ``fastparquet`` otherwise.
+        it is installed, and falls back to the deprecated ``fastparquet`` otherwise.
 
-    Returns
-    -------
-    A dict containing a ``'read'`` and ``'write'`` function.
+    This can be used to inject third-party engine; e.g. from dask_cudf.
     """
-    if engine in _ENGINES:
-        return _ENGINES[engine]
+    if isinstance(engine, type) and issubclass(engine, Engine):
+        return engine
 
-    if engine == "auto":
-        try:
-            return get_engine("pyarrow")
-        except RuntimeError:
-            pass
+    if engine in ("arrow", "pyarrow-dataset"):
+        warnings.warn(
+            f"engine='{engine}' has been deprecated. "
+            "Please use engine='pyarrow' instead.",
+            category=FutureWarning,
+        )
+        engine = "pyarrow"
+
+    if engine in ("auto", "pyarrow"):
+        if "pyarrow" in _ENGINES:
+            return _ENGINES["pyarrow"]
 
         try:
-            return get_engine("fastparquet")
+            import_required("pyarrow", "`pyarrow` not installed")
         except RuntimeError:
-            raise RuntimeError("Please install either pyarrow or fastparquet") from None
+            if engine != "auto":
+                raise
+        else:
+            from dask.dataframe.io.parquet.arrow import ArrowDatasetEngine
 
-    elif engine == "fastparquet":
-        import_required("fastparquet", "`fastparquet` not installed")
-        from dask.dataframe.io.parquet.fastparquet import FastParquetEngine
+            _ENGINES["pyarrow"] = eng = ArrowDatasetEngine
+            return eng
 
-        _ENGINES["fastparquet"] = eng = FastParquetEngine
-        return eng
+    if engine in ("auto", "fastparquet"):
 
-    elif engine in ("pyarrow", "arrow", "pyarrow-dataset"):
-        import_required("pyarrow", "`pyarrow` not installed")
+        def warn_fastparquet():
+            if engine == "fastparquet":
+                warnings.warn(
+                    "The fastparquet engine is deprecated and will be removed "
+                    "in a future release. Please install pyarrow.",
+                    category=FutureWarning,
+                )
+            else:
+                warnings.warn(
+                    "Could not find pyarrow; falling back to fastparquet, which "
+                    "is deprecated and will be removed in a future release.",
+                    category=FutureWarning,
+                )
 
-        if engine in ("pyarrow-dataset", "arrow"):
-            engine = "pyarrow"
+        if "fastparquet" in _ENGINES:
+            warn_fastparquet()
+            return _ENGINES["fastparquet"]
 
-        if engine == "pyarrow":
-            from dask.dataframe.io.parquet.arrow import ArrowDatasetEngine
+        try:
+            import_required("fastparquet", "`fastparquet` not installed")
+        except RuntimeError:
+            if engine == "auto":
+                raise RuntimeError("`pyarrow` not installed")
+            else:
+                raise
+        else:
+            from dask.dataframe.io.parquet.fastparquet import FastParquetEngine
 
-            _ENGINES[engine] = eng = ArrowDatasetEngine
-        return eng
+            _ENGINES["fastparquet"] = fpq_eng = FastParquetEngine
+            warn_fastparquet()
+            return fpq_eng
 
-    else:
-        raise ValueError(
-            f'Unsupported engine: "{engine}".'
-            '  Valid choices include "pyarrow" and "fastparquet".'
-        )
+    raise ValueError(
+        f'Unsupported engine: "{engine}". Valid choices are "pyarrow" or "auto".'
+    )
 
 
 #####################
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/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..9e8a8beed 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
 
@@ -2061,7 +2185,6 @@ def show_versions() -> None:
         "cloudpickle",
         "fsspec",
         "bokeh",
-        "fastparquet",
         "pyarrow",
         "zarr",
     ]
diff --git a/docs/source/dataframe-hive.rst b/docs/source/dataframe-hive.rst
index 8b8c2ba80..7859a193a 100644
--- a/docs/source/dataframe-hive.rst
+++ b/docs/source/dataframe-hive.rst
@@ -118,9 +118,8 @@ be interpreted as categorical columns.
 Defining a Custom Partitioning Schema
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-When utilizing ``engine='pyarrow'``, it is possible to specify a custom
-schema for the hive-partitioned columns. The columns will then be read
-using the specified types and not as `category`.
+It is possible to specify a custom schema for the hive-partitioned columns. The columns
+will then be read using the specified types and not as `category`.
 
 .. code-block:: python
 
diff --git a/docs/source/dataframe-parquet.rst b/docs/source/dataframe-parquet.rst
index 7d92d5029..a271d5391 100644
--- a/docs/source/dataframe-parquet.rst
+++ b/docs/source/dataframe-parquet.rst
@@ -11,6 +11,9 @@ and retrieval. Dask dataframe includes :func:`read_parquet` and
 respectively. Here we document these methods, and provide some tips and best
 practices.
 
+Parquet I/O requires ``pyarrow`` to be installed.
+
+
 Reading Parquet Files
 ---------------------
 
@@ -61,21 +64,6 @@ For more information on connecting to remote data, see
 :func:`read_parquet` has many configuration options affecting both behavior and
 performance. Here we highlight a few common options.
 
-Engine
-~~~~~~
-
-:func:`read_parquet` supports two backend engines - ``pyarrow`` and
-``fastparquet``. The ``pyarrow`` engine is used by default, falling back to
-``fastparquet`` if ``pyarrow`` isn't installed. If desired, you may explicitly
-specify the engine using the ``engine`` keyword argument:
-
-.. code-block:: python
-
-   >>> df = dd.read_parquet(
-   ...      "s3://bucket-name/my/parquet/",
-   ...      engine="fastparquet"  # explicitly specify the fastparquet engine
-   ... )
-
 Metadata
 ~~~~~~~~
 
@@ -224,21 +212,6 @@ your data is partitioned optimally.
 :func:`to_parquet` has many configuration options affecting both behavior and
 performance. Here we highlight a few common options.
 
-Engine
-~~~~~~
-
-:func:`to_parquet` supports two backend engines - ``pyarrow`` and
-``fastparquet``. The ``pyarrow`` engine is used by default, falling back to
-``fastparquet`` if ``pyarrow`` isn't installed. If desired, you may explicitly
-specify the engine using the ``engine`` keyword argument:
-
-.. code-block:: python
-
-   >>> df.to_parquet(
-   ...      "s3://bucket-name/my/parquet/",
-   ...      engine="fastparquet"  # explicitly specify the fastparquet engine
-   ... )
-
 Metadata
 ~~~~~~~~
 
diff --git a/docs/source/install.rst b/docs/source/install.rst
index c3e2a340f..8a0090c86 100644
--- a/docs/source/install.rst
+++ b/docs/source/install.rst
@@ -118,8 +118,6 @@ These optional dependencies and their minimum supported versions are listed belo
 +------------------+-----------------+---------------------------------------------------------------------------------------------------------+
 | `fastavro`_      | ``>=1.1.0``     | Storing and reading data from Apache Avro files                                                         |
 +------------------+-----------------+---------------------------------------------------------------------------------------------------------+
-| `fastparquet`_   | ``>=0.8.2``     | Storing and reading data from Apache Parquet files                                                      |
-+------------------+-----------------+---------------------------------------------------------------------------------------------------------+
 | `gcsfs`_         | ``>=2021.9.0``  | Storing and reading data located in Google Cloud Storage                                                |
 +------------------+-----------------+---------------------------------------------------------------------------------------------------------+
 | `graphviz`_      | ``>=0.8.4``     | Graph visualization using the graphviz engine                                                           |
@@ -199,7 +197,6 @@ See the :ref:`section on testing <develop-test>` in the Development Guidelines f
 .. _cytoolz: https://github.com/pytoolz/cytoolz
 .. _dask-ml: https://ml.dask.org/
 .. _fastavro: https://fastavro.readthedocs.io/en/latest/
-.. _fastparquet: https://github.com/dask/fastparquet
 .. _graphviz: https://graphviz.readthedocs.io/en/stable/
 .. _gcsfs: https://gcsfs.readthedocs.io/en/latest/
 .. _h5py: https://www.h5py.org/


Test patch:
diff --git a/dask/bytes/tests/test_http.py b/dask/bytes/tests/test_http.py
index 6be8ecdb4..792704973 100644
--- a/dask/bytes/tests/test_http.py
+++ b/dask/bytes/tests/test_http.py
@@ -171,7 +171,15 @@ def test_open_glob(dir_server):
 
 
 @pytest.mark.network
-@pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
+@pytest.mark.parametrize(
+    "engine",
+    [
+        "pyarrow",
+        pytest.param(
+            "fastparquet", marks=pytest.mark.filterwarnings("ignore::FutureWarning")
+        ),
+    ],
+)
 def test_parquet(engine):
     pytest.importorskip("requests", minversion="2.21.0")
     dd = pytest.importorskip("dask.dataframe")
diff --git a/dask/bytes/tests/test_s3.py b/dask/bytes/tests/test_s3.py
index 4a005ec6d..ddbb7ad86 100644
--- a/dask/bytes/tests/test_s3.py
+++ b/dask/bytes/tests/test_s3.py
@@ -439,13 +439,24 @@ def test_modification_time_read_bytes(s3, s3so):
     assert [aa._key for aa in concat(a)] != [cc._key for cc in concat(c)]
 
 
-@pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
+@pytest.fixture(
+    params=[
+        "pyarrow",
+        pytest.param(
+            "fastparquet", marks=pytest.mark.filterwarnings("ignore::FutureWarning")
+        ),
+    ]
+)
+def engine(request):
+    pytest.importorskip(request.param)
+    return request.param
+
+
 @pytest.mark.parametrize("metadata_file", [True, False])
 def test_parquet(s3, engine, s3so, metadata_file):
     dd = pytest.importorskip("dask.dataframe")
     pd = pytest.importorskip("pandas")
     np = pytest.importorskip("numpy")
-    pytest.importorskip(engine)
 
     url = "s3://%s/test.parquet" % test_bucket_name
     data = pd.DataFrame(
@@ -538,9 +549,7 @@ def test_parquet(s3, engine, s3so, metadata_file):
     dd.utils.assert_eq(data, df4)
 
 
-@pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
 def test_parquet_append(s3, engine, s3so):
-    pytest.importorskip(engine)
     dd = pytest.importorskip("dask.dataframe")
     pd = pytest.importorskip("pandas")
     np = pytest.importorskip("numpy")
@@ -593,9 +602,7 @@ def test_parquet_append(s3, engine, s3so):
     )
 
 
-@pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
 def test_parquet_wstoragepars(s3, s3so, engine):
-    pytest.importorskip(engine)
     dd = pytest.importorskip("dask.dataframe")
     pd = pytest.importorskip("pandas")
     np = pytest.importorskip("numpy")
@@ -615,18 +622,20 @@ def test_parquet_wstoragepars(s3, s3so, engine):
     dd.read_parquet(
         url,
         engine=engine,
-        storage_options=dict(**s3so, **{"default_fill_cache": False}),
+        storage_options={"default_fill_cache": False, **s3so},
     )
     assert s3.current().default_fill_cache is False
     dd.read_parquet(
-        url, engine=engine, storage_options=dict(**s3so, **{"default_fill_cache": True})
+        url,
+        engine=engine,
+        storage_options={"default_fill_cache": True, **s3so},
     )
     assert s3.current().default_fill_cache is True
 
     dd.read_parquet(
         url,
         engine=engine,
-        storage_options=dict(**s3so, **{"default_block_size": 2**20}),
+        storage_options={"default_block_size": 2**20, **s3so},
     )
     assert s3.current().default_block_size == 2**20
     with s3.current().open(url + "/_metadata") as f:
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 3b0d63a0a..a68ebfdd6 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -72,70 +72,88 @@ df = pd.DataFrame(
 
 ddf = dd.from_pandas(df, npartitions=npartitions)
 
-
-@pytest.fixture(
+_engine_fixture = pytest.fixture(
     params=[
-        pytest.param("fastparquet", marks=FASTPARQUET_MARK),
-        pytest.param("pyarrow", marks=PYARROW_MARK),
+        pytest.param(
+            "fastparquet",
+            marks=[
+                FASTPARQUET_MARK,
+                pytest.mark.filterwarnings("ignore::FutureWarning"),
+            ],
+        ),
+        pytest.param("pyarrow", marks=[PYARROW_MARK]),
     ]
 )
-def engine(request):
-    return request.param
-
 
-def write_read_engines(**kwargs):
-    """Product of both engines for write/read:
 
-    To add custom marks, pass keyword of the form: `mark_writer_reader=reason`,
-    or `mark_engine=reason` to apply to all parameters with that engine."""
-    backends = {"pyarrow", "fastparquet"}
+@_engine_fixture
+def engine(request):
+    return request.param
 
-    # Skip if uninstalled
-    skip_marks = {
-        "fastparquet": FASTPARQUET_MARK,
-        "pyarrow": PYARROW_MARK,
-    }
-    marks = {(w, r): [skip_marks[w], skip_marks[r]] for w in backends for r in backends}
-
-    # Custom marks
-    for kw, val in kwargs.items():
-        kind, rest = kw.split("_", 1)
-        key = tuple(rest.split("_"))
-        if kind not in ("xfail", "skip") or len(key) > 2 or set(key) - backends:
-            raise ValueError("unknown keyword %r" % kw)
-        val = getattr(pytest.mark, kind)(reason=val)
-        if len(key) == 2:
-            marks[key].append(val)
-        else:
-            for k in marks:
-                if key in k:
-                    marks[k].append(val)
 
-    return pytest.mark.parametrize(
-        ("write_engine", "read_engine"),
-        [pytest.param(*k, marks=tuple(v)) for (k, v) in sorted(marks.items())],
-    )
+@_engine_fixture
+def write_engine(request):
+    return request.param
 
 
-fp_pandas_xfail = write_read_engines()
+@_engine_fixture
+def read_engine(request):
+    return request.param
 
 
 @PYARROW_MARK
 def test_get_engine_pyarrow():
     from dask.dataframe.io.parquet.arrow import ArrowDatasetEngine
 
-    assert get_engine("pyarrow") == ArrowDatasetEngine
-    assert get_engine("arrow") == ArrowDatasetEngine
+    assert get_engine("auto") is ArrowDatasetEngine
+    assert get_engine("pyarrow") is ArrowDatasetEngine
+    with pytest.warns(FutureWarning):
+        assert get_engine("pyarrow-dataset") is ArrowDatasetEngine
+    with pytest.warns(FutureWarning):
+        assert get_engine("arrow") is ArrowDatasetEngine
 
 
 @FASTPARQUET_MARK
 def test_get_engine_fastparquet():
     from dask.dataframe.io.parquet.fastparquet import FastParquetEngine
 
-    assert get_engine("fastparquet") == FastParquetEngine
+    with pytest.warns(FutureWarning):
+        assert get_engine("fastparquet") is FastParquetEngine
+
+
+@pytest.mark.skipif(
+    pa or not fastparquet, reason="Must have fastparquet but not pyarrow installed"
+)
+def test_get_engine_fastparquet_only():
+    from dask.dataframe.io.parquet.fastparquet import FastParquetEngine
+
+    with pytest.warns(
+        FutureWarning, match="Could not find pyarrow; falling back to fastparquet"
+    ):
+        assert get_engine("auto") is FastParquetEngine
+
+
+@pytest.mark.skipif(pa or fastparquet, reason="fastparquet or pyarrow are installed")
+def test_get_engine_no_engine():
+    with pytest.raises(RuntimeError, match="`pyarrow` not installed"):
+        get_engine("auto")
+
+
+def test_get_engine_third_party():
+    """Test extensibility by dask_cudf"""
+    from dask.dataframe.io.parquet.utils import Engine
+
+    class MyEngine(Engine):
+        pass
+
+    assert get_engine(MyEngine) is MyEngine
+
+
+def test_get_engine_invalid():
+    with pytest.raises(ValueError):
+        get_engine(123)
 
 
-@write_read_engines()
 @pytest.mark.parametrize("has_metadata", [False, True])
 def test_local(tmpdir, write_engine, read_engine, has_metadata):
     tmp = str(tmpdir)
@@ -170,7 +188,6 @@ def test_local(tmpdir, write_engine, read_engine, has_metadata):
 
 
 @pytest.mark.parametrize("index", [False, True])
-@write_read_engines()
 def test_empty(tmpdir, write_engine, read_engine, index):
     fn = str(tmpdir)
     df = pd.DataFrame({"a": ["a", "b", "b"], "b": [4, 5, 6]})[:0]
@@ -183,7 +200,6 @@ def test_empty(tmpdir, write_engine, read_engine, index):
     assert_eq(ddf, read_df)
 
 
-@write_read_engines()
 def test_simple(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
     df = pd.DataFrame({"a": ["a", "b", "b"], "b": [4, 5, 6]})
@@ -196,7 +212,6 @@ def test_simple(tmpdir, write_engine, read_engine):
     assert_eq(ddf, read_df)
 
 
-@write_read_engines()
 def test_delayed_no_metadata(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
     df = pd.DataFrame({"a": ["a", "b", "b"], "b": [4, 5, 6]})
@@ -217,7 +232,6 @@ def test_delayed_no_metadata(tmpdir, write_engine, read_engine):
     assert_eq(ddf, read_df)
 
 
-@write_read_engines()
 def test_read_glob(tmpdir, write_engine, read_engine):
     tmp_path = str(tmpdir)
     ddf.to_parquet(tmp_path, engine=write_engine)
@@ -235,7 +249,6 @@ def test_read_glob(tmpdir, write_engine, read_engine):
     assert_eq(ddf, ddf2)
 
 
-@write_read_engines()
 def test_calculate_divisions_false(tmpdir, write_engine, read_engine):
     tmp_path = str(tmpdir)
     ddf.to_parquet(tmp_path, write_index=False, engine=write_engine)
@@ -249,7 +262,6 @@ def test_calculate_divisions_false(tmpdir, write_engine, read_engine):
     assert_eq(ddf, ddf2, check_index=False, check_divisions=False)
 
 
-@write_read_engines()
 def test_read_list(tmpdir, write_engine, read_engine):
     if write_engine == read_engine == "fastparquet" and os.name == "nt":
         # fastparquet or dask is not normalizing filepaths correctly on
@@ -273,7 +285,6 @@ def test_read_list(tmpdir, write_engine, read_engine):
     assert_eq(ddf, ddf2)
 
 
-@write_read_engines()
 def test_columns_auto_index(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
     ddf.to_parquet(fn, engine=write_engine)
@@ -311,7 +322,6 @@ def test_columns_auto_index(tmpdir, write_engine, read_engine):
     )
 
 
-@write_read_engines()
 def test_columns_index(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
     ddf.to_parquet(fn, engine=write_engine)
@@ -401,12 +411,11 @@ def test_nonsense_column(tmpdir, engine):
     fn = str(tmpdir)
     ddf.to_parquet(fn, engine=engine)
     with pytest.raises((ValueError, KeyError)):
-        dd.read_parquet(fn, columns=["nonesense"], engine=engine)
+        dd.read_parquet(fn, columns=["nonsense"], engine=engine)
     with pytest.raises((Exception, KeyError)):
-        dd.read_parquet(fn, columns=["nonesense"] + list(ddf.columns), engine=engine)
+        dd.read_parquet(fn, columns=["nonsense"] + list(ddf.columns), engine=engine)
 
 
-@write_read_engines()
 def test_columns_no_index(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
     ddf.to_parquet(fn, engine=write_engine)
@@ -451,7 +460,6 @@ def test_columns_no_index(tmpdir, write_engine, read_engine):
     )
 
 
-@write_read_engines()
 def test_calculate_divisions_no_index(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
     ddf.to_parquet(fn, engine=write_engine, write_index=False)
@@ -523,7 +531,6 @@ def test_columns_index_with_multi_index(tmpdir, engine):
         assert_eq(d, sol_df[col])
 
 
-@write_read_engines()
 def test_no_index(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
     df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
@@ -561,27 +568,22 @@ def test_names(tmpdir, engine):
     assert set(read(fn, columns=("x",)).dask) == set(read(fn, columns=["x"]).dask)
 
 
-@write_read_engines()
 def test_roundtrip_from_pandas(tmpdir, write_engine, read_engine):
     fn = str(tmpdir.join("test.parquet"))
     dfp = df.copy()
     dfp.index.name = "index"
-    dfp.to_parquet(
-        fn, engine="pyarrow" if write_engine.startswith("pyarrow") else "fastparquet"
-    )
+    dfp.to_parquet(fn, engine=write_engine)
     ddf = dd.read_parquet(fn, index="index", engine=read_engine)
     assert_eq(dfp, ddf)
 
 
-@write_read_engines()
-def test_roundtrip_nullable_dtypes(tmp_path, write_engine, read_engine):
-    """
-    Test round-tripping nullable extension dtypes. Parquet engines will
+@PYARROW_MARK
+def test_roundtrip_nullable_dtypes(tmp_path):
+    """Test round-tripping nullable extension dtypes. Parquet engines will
     typically add dtype metadata for this.
-    """
-    if read_engine == "fastparquet" or write_engine == "fastparquet":
-        pytest.xfail("https://github.com/dask/fastparquet/issues/465")
 
+    fastparquet is unsupported: https://github.com/dask/fastparquet/issues/465
+    """
     df = pd.DataFrame(
         {
             "a": pd.Series([1, 2, pd.NA, 3, 4], dtype="Int64"),
@@ -591,8 +593,8 @@ def test_roundtrip_nullable_dtypes(tmp_path, write_engine, read_engine):
         }
     )
     ddf = dd.from_pandas(df, npartitions=2)
-    ddf.to_parquet(tmp_path, engine=write_engine)
-    ddf2 = dd.read_parquet(tmp_path, engine=read_engine)
+    ddf.to_parquet(tmp_path)
+    ddf2 = dd.read_parquet(tmp_path)
     assert_eq(df, ddf2)
 
 
@@ -691,7 +693,6 @@ def test_use_nullable_dtypes_with_types_mapper(tmp_path, engine):
     assert_eq(result, expected)
 
 
-@write_read_engines()
 def test_categorical(tmpdir, write_engine, read_engine):
     if write_engine == "fastparquet" and read_engine == "pyarrow":
         pytest.xfail("Known limitation")
@@ -855,8 +856,8 @@ def test_partition_on_cats_pyarrow(tmpdir, stats, meta):
         }
     )
     d = dd.from_pandas(d, 2)
-    d.to_parquet(tmp, partition_on=["b"], engine="pyarrow", write_metadata_file=meta)
-    df = dd.read_parquet(tmp, engine="pyarrow", calculate_divisions=stats)
+    d.to_parquet(tmp, partition_on=["b"], write_metadata_file=meta)
+    df = dd.read_parquet(tmp, calculate_divisions=stats)
     assert set(df.b.cat.categories) == {"x", "y", "z"}
 
 
@@ -1028,13 +1029,12 @@ def test_append_different_columns(tmpdir, engine, metadata_file):
     assert "Appended dtypes" in str(excinfo.value)
 
 
+@PYARROW_MARK
 @pytest.mark.skip_with_pyarrow_strings  # need an object to store a dict
-def test_append_dict_column(tmpdir, engine):
-    # See: https://github.com/dask/dask/issues/7492
-
-    if engine == "fastparquet":
-        pytest.xfail("Fastparquet engine is missing dict-column support")
-
+def test_append_dict_column(tmpdir):
+    """See: https://github.com/dask/dask/issues/7492
+    Note: fastparquet engine is missing dict-column support
+    """
     tmp = str(tmpdir)
     dts = pd.date_range("2020-01-01", "2021-01-01")
     df = pd.DataFrame(
@@ -1046,13 +1046,11 @@ def test_append_dict_column(tmpdir, engine):
     schema = {"value": pa.struct([("x", pa.int32())])}
 
     # Write ddf1 to tmp, and then append it again
-    ddf1.to_parquet(tmp, append=True, engine=engine, schema=schema)
-    ddf1.to_parquet(
-        tmp, append=True, engine=engine, schema=schema, ignore_divisions=True
-    )
+    ddf1.to_parquet(tmp, append=True, schema=schema)
+    ddf1.to_parquet(tmp, append=True, schema=schema, ignore_divisions=True)
 
     # Read back all data (ddf1 + ddf1)
-    ddf2 = dd.read_parquet(tmp, engine=engine)
+    ddf2 = dd.read_parquet(tmp)
 
     # Check computed result
     expect = pd.concat([df, df])
@@ -1060,7 +1058,6 @@ def test_append_dict_column(tmpdir, engine):
     assert_eq(expect, result)
 
 
-@write_read_engines()
 def test_ordering(tmpdir, write_engine, read_engine):
     tmp = str(tmpdir)
     df = pd.DataFrame(
@@ -1139,7 +1136,7 @@ def test_read_parquet_custom_columns(tmpdir, engine):
 )
 @pytest.mark.skip_with_pyarrow_strings  # don't want to convert binary data to pyarrow strings
 def test_roundtrip(tmpdir, df, write_kwargs, read_kwargs, engine):
-    if "x" in df and df.x.dtype == "M8[ns]" and "arrow" in engine:
+    if "x" in df and df.x.dtype == "M8[ns]" and engine == "pyarrow":
         pytest.xfail(reason="Parquet pyarrow v1 doesn't support nanosecond precision")
     if (
         "x" in df
@@ -1161,13 +1158,6 @@ def test_roundtrip(tmpdir, df, write_kwargs, read_kwargs, engine):
         ):
             pytest.xfail(reason="https://github.com/dask/fastparquet/issues/837")
 
-    if (
-        read_kwargs.get("categories", None)
-        and engine == "fastparquet"
-        and fastparquet_version <= parse_version("0.6.3")
-    ):
-        pytest.xfail("https://github.com/dask/fastparquet/issues/577")
-
     tmp = str(tmpdir)
     if df.index.name is None:
         df.index.name = "index"
@@ -1284,7 +1274,8 @@ def test_to_parquet_fastparquet_default_writes_nulls(tmpdir):
     df = pd.DataFrame({"c1": [1.0, np.nan, 2, np.nan, 3]})
     ddf = dd.from_pandas(df, npartitions=1)
 
-    ddf.to_parquet(fn, engine="fastparquet")
+    with pytest.warns(FutureWarning):
+        ddf.to_parquet(fn, engine="fastparquet")
     table = pq.read_table(fn)
     assert table[1].null_count == 2
 
@@ -1339,11 +1330,9 @@ def test_to_parquet_pyarrow_w_inconsistent_schema_by_partition_succeeds_w_manual
             ("partition_column", pa.int64()),
         ]
     )
-    ddf.to_parquet(
-        str(tmpdir), engine="pyarrow", partition_on="partition_column", schema=schema
-    )
+    ddf.to_parquet(str(tmpdir), partition_on="partition_column", schema=schema)
     ddf_after_write = (
-        dd.read_parquet(str(tmpdir), engine="pyarrow", calculate_divisions=False)
+        dd.read_parquet(str(tmpdir), calculate_divisions=False)
         .compute()
         .reset_index(drop=True)
     )
@@ -1409,8 +1398,8 @@ def test_pyarrow_schema_inference(tmpdir, index, schema):
     else:
         df = dd.from_pandas(df, npartitions=2)
 
-    df.to_parquet(tmpdir, engine="pyarrow", schema=schema)
-    df_out = dd.read_parquet(tmpdir, engine="pyarrow", calculate_divisions=True)
+    df.to_parquet(tmpdir, schema=schema)
+    df_out = dd.read_parquet(tmpdir, calculate_divisions=True)
     assert_eq(df, df_out)
 
 
@@ -1424,7 +1413,7 @@ def test_pyarrow_schema_mismatch_error(tmpdir):
     )
 
     with pytest.raises(ValueError) as rec:
-        ddf.to_parquet(str(tmpdir), engine="pyarrow")
+        ddf.to_parquet(str(tmpdir))
 
     msg = str(rec.value)
     assert "Failed to convert partition to expected pyarrow schema" in msg
@@ -1439,8 +1428,8 @@ def test_pyarrow_schema_mismatch_explicit_schema_none(tmpdir):
     ddf = dd.from_delayed(
         [dask.delayed(df1), dask.delayed(df2)], meta=df1, verify_meta=False
     )
-    ddf.to_parquet(str(tmpdir), engine="pyarrow", schema=None)
-    res = dd.read_parquet(tmpdir, engine="pyarrow")
+    ddf.to_parquet(str(tmpdir), schema=None)
+    res = dd.read_parquet(tmpdir)
     sol = pd.concat([df1, df2])
     # Only checking that the data was written correctly, we don't care about
     # the incorrect _meta from read_parquet
@@ -1516,18 +1505,14 @@ def test_partition_on_string(tmpdir, partition_on):
             }
         )
         d = dd.from_pandas(df, npartitions=2)
-        d.to_parquet(
-            tmpdir, partition_on=partition_on, write_index=False, engine="pyarrow"
-        )
-        out = dd.read_parquet(
-            tmpdir, index=False, calculate_divisions=False, engine="pyarrow"
-        )
+        d.to_parquet(tmpdir, partition_on=partition_on, write_index=False)
+        out = dd.read_parquet(tmpdir, index=False, calculate_divisions=False)
+
     out = out.compute()
     for val in df.aa.unique():
         assert set(df.bb[df.aa == val]) == set(out.bb[out.aa == val])
 
 
-@write_read_engines()
 def test_filters_categorical(tmpdir, write_engine, read_engine):
     tmpdir = str(tmpdir)
     cats = ["2018-01-01", "2018-01-02", "2018-01-03", "2018-01-04"]
@@ -1549,7 +1534,6 @@ def test_filters_categorical(tmpdir, write_engine, read_engine):
     assert len(ddftest_read) == 2
 
 
-@write_read_engines()
 def test_filters(tmpdir, write_engine, read_engine):
     tmp_path = str(tmpdir)
     df = pd.DataFrame({"x": range(10), "y": list("aabbccddee")})
@@ -1598,15 +1582,7 @@ def test_filters(tmpdir, write_engine, read_engine):
     assert g.npartitions == 5
 
 
-@write_read_engines()
 def test_filters_v0(tmpdir, write_engine, read_engine):
-    if write_engine == "fastparquet" or read_engine == "fastparquet":
-        pytest.importorskip("fastparquet", minversion="0.3.1")
-
-    # Recent versions of pyarrow support full row-wise filtering
-    # (fastparquet and older pyarrow versions do not)
-    pyarrow_row_filtering = read_engine == "pyarrow"
-
     fn = str(tmpdir)
     df = pd.DataFrame({"at": ["ab", "aa", "ba", "da", "bb"]})
     ddf = dd.from_pandas(df, npartitions=1)
@@ -1621,7 +1597,9 @@ def test_filters_v0(tmpdir, write_engine, read_engine):
     ddf3 = dd.read_parquet(
         fn, index=False, engine=read_engine, filters=[("at", "=", "aa")]
     ).compute()
-    if pyarrow_row_filtering:
+
+    # Pyarrow supports full row-wise filtering; fastparquet does not.
+    if read_engine == "pyarrow":
         assert_eq(ddf2, ddf[ddf["at"] == "aa"], check_index=False)
         assert_eq(ddf3, ddf[ddf["at"] == "aa"], check_index=False)
     else:
@@ -1654,9 +1632,8 @@ def test_filters_v0(tmpdir, write_engine, read_engine):
     assert_eq(ddf2, ddf3)
 
 
+@PYARROW_MARK
 def test_filtering_pyarrow_dataset(tmpdir, engine):
-    pytest.importorskip("pyarrow", minversion="1.0.0")
-
     fn = str(tmpdir)
     df = pd.DataFrame({"aa": range(100), "bb": ["cat", "dog"] * 50})
     ddf = dd.from_pandas(df, npartitions=10)
@@ -1716,27 +1693,21 @@ def test_filters_file_list(tmpdir, engine):
     assert_eq(df[df["x"] > 3], ddf3, check_index=False)
 
 
+@PYARROW_MARK
 def test_pyarrow_filter_divisions(tmpdir):
-    pytest.importorskip("pyarrow")
-
     # Write simple dataset with an index that will only
     # have a sorted index if certain row-groups are filtered out.
     # In this case, we filter "a" <= 3 to get a sorted
     # index. Otherwise, "a" is NOT monotonically increasing.
     df = pd.DataFrame({"a": [0, 1, 10, 12, 2, 3, 8, 9], "b": range(8)}).set_index("a")
-    df.iloc[:4].to_parquet(
-        str(tmpdir.join("file.0.parquet")), engine="pyarrow", row_group_size=2
-    )
-    df.iloc[4:].to_parquet(
-        str(tmpdir.join("file.1.parquet")), engine="pyarrow", row_group_size=2
-    )
+    df.iloc[:4].to_parquet(str(tmpdir.join("file.0.parquet")), row_group_size=2)
+    df.iloc[4:].to_parquet(str(tmpdir.join("file.1.parquet")), row_group_size=2)
 
     # Only works for ArrowDatasetEngine.
     # Legacy code will not apply filters on individual row-groups
     # when `split_row_groups=False`.
     ddf = dd.read_parquet(
         str(tmpdir),
-        engine="pyarrow",
         split_row_groups=False,
         calculate_divisions=True,
         filters=[("a", "<=", 3)],
@@ -1745,7 +1716,6 @@ def test_pyarrow_filter_divisions(tmpdir):
 
     ddf = dd.read_parquet(
         str(tmpdir),
-        engine="pyarrow",
         split_row_groups=True,
         calculate_divisions=True,
         filters=[("a", "<=", 3)],
@@ -1754,8 +1724,8 @@ def test_pyarrow_filter_divisions(tmpdir):
 
 
 @FASTPARQUET_MARK
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 def test_divisions_read_with_filters(tmpdir):
-    pytest.importorskip("fastparquet", minversion="0.3.1")
     tmpdir = str(tmpdir)
     # generate dataframe
     size = 100
@@ -1785,8 +1755,8 @@ def test_divisions_read_with_filters(tmpdir):
 
 
 @FASTPARQUET_MARK
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 def test_divisions_are_known_read_with_filters(tmpdir):
-    pytest.importorskip("fastparquet", minversion="0.3.1")
     tmpdir = str(tmpdir)
     # generate dataframe
     df = pd.DataFrame(
@@ -1812,37 +1782,6 @@ def test_divisions_are_known_read_with_filters(tmpdir):
     assert out.divisions == expected_divisions
 
 
-@FASTPARQUET_MARK
-@pytest.mark.xfail(reason="No longer accept ParquetFile objects")
-def test_read_from_fastparquet_parquetfile(tmpdir):
-    fn = str(tmpdir)
-
-    df = pd.DataFrame(
-        {
-            "a": np.random.choice(["A", "B", "C"], size=100),
-            "b": np.random.random(size=100),
-            "c": np.random.randint(1, 5, size=100),
-        }
-    )
-    d = dd.from_pandas(df, npartitions=2)
-    d.to_parquet(fn, partition_on=["a"], engine="fastparquet")
-
-    pq_f = fastparquet.ParquetFile(fn)
-
-    # OK with no filters
-    out = dd.read_parquet(pq_f).compute()
-    for val in df.a.unique():
-        assert set(df.b[df.a == val]) == set(out.b[out.a == val])
-
-    # OK with  filters
-    out = dd.read_parquet(pq_f, filters=[("a", "==", "B")]).compute()
-    assert set(df.b[df.a == "B"]) == set(out.b)
-
-    # Engine should not be set to 'pyarrow'
-    with pytest.raises(AssertionError):
-        out = dd.read_parquet(pq_f, engine="pyarrow")
-
-
 @pytest.mark.parametrize("scheduler", ["threads", "processes"])
 def test_to_parquet_lazy(tmpdir, scheduler, engine):
     tmpdir = str(tmpdir)
@@ -1867,13 +1806,14 @@ def test_to_parquet_calls_invalidate_cache(tmpdir, monkeypatch, compute):
 
     invalidate_cache = MagicMock()
     monkeypatch.setattr(LocalFileSystem, "invalidate_cache", invalidate_cache)
-    ddf.to_parquet(tmpdir, compute=compute, engine="pyarrow")
+    ddf.to_parquet(tmpdir, compute=compute)
     path = LocalFileSystem._strip_protocol(str(tmpdir))
     assert invalidate_cache.called
     assert invalidate_cache.call_args.args[0] == path
 
 
 @FASTPARQUET_MARK
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 def test_timestamp96(tmpdir):
     fn = str(tmpdir)
     df = pd.DataFrame({"a": [pd.to_datetime("now", utc=True)]})
@@ -1886,6 +1826,7 @@ def test_timestamp96(tmpdir):
 
 
 @FASTPARQUET_MARK
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 def test_drill_scheme(tmpdir):
     fn = str(tmpdir)
     N = 5
@@ -1930,8 +1871,6 @@ def test_parquet_select_cats(tmpdir, engine):
 
 
 def test_columns_name(tmpdir, engine):
-    if engine == "fastparquet" and fastparquet_version <= parse_version("0.3.1"):
-        pytest.skip("Fastparquet does not write column_indexes up to 0.3.1")
     tmp_path = str(tmpdir)
     df = pd.DataFrame({"A": [1, 2]}, index=pd.Index(["a", "b"], name="idx"))
     df.columns.name = "cols"
@@ -1951,6 +1890,7 @@ def check_compression(engine, filename, compression):
         else:
             assert md.total_compressed_size != md.total_uncompressed_size
     else:
+        assert engine == "pyarrow"
         metadata = pa.parquet.read_metadata(os.path.join(filename, "_metadata"))
         names = metadata.schema.names
         for i in range(metadata.num_row_groups):
@@ -2279,24 +2219,26 @@ def test_writing_parquet_with_kwargs(tmpdir, engine):
     df.index.name = "index"
     ddf = dd.from_pandas(df, npartitions=3)
 
-    engine_kwargs = {
+    write_kwargs = {
         "pyarrow": {
             "compression": "snappy",
             "coerce_timestamps": None,
             "use_dictionary": True,
         },
-        "fastparquet": {"compression": "snappy", "times": "int64", "fixed_text": None},
+        "fastparquet": {
+            "compression": "snappy",
+            "times": "int64",
+            "fixed_text": None,
+        },
     }
 
-    ddf.to_parquet(path1, engine=engine, **engine_kwargs[engine])
+    ddf.to_parquet(path1, engine=engine, **write_kwargs[engine])
     out = dd.read_parquet(path1, engine=engine, calculate_divisions=True)
     assert_eq(out, ddf, check_index=(engine != "fastparquet"))
 
     # Avoid race condition in pyarrow 0.8.0 on writing partitioned datasets
     with dask.config.set(scheduler="sync"):
-        ddf.to_parquet(
-            path2, engine=engine, partition_on=["a"], **engine_kwargs[engine]
-        )
+        ddf.to_parquet(path2, engine=engine, partition_on=["a"], **write_kwargs[engine])
     out = dd.read_parquet(path2, engine=engine).compute()
     for val in df.a.unique():
         assert set(df.b[df.a == val]) == set(out.b[out.a == val])
@@ -2351,20 +2293,17 @@ def test_select_partitioned_column(tmpdir, engine):
     df_partitioned[df_partitioned.fake_categorical1 == "A"].compute()
 
 
-def test_with_tz(tmpdir, engine):
-    if engine == "fastparquet" and fastparquet_version < parse_version("0.3.0"):
-        pytest.skip("fastparquet<0.3.0 did not support this")
-
+@FASTPARQUET_MARK
+def test_with_tz(tmpdir):
     with warnings.catch_warnings():
-        if engine == "fastparquet":
-            # fastparquet-442
-            warnings.simplefilter("ignore", FutureWarning)  # pandas 0.25
-            fn = str(tmpdir)
-            df = pd.DataFrame([[0]], columns=["a"], dtype="datetime64[ns, UTC]")
-            df = dd.from_pandas(df, 1)
-            df.to_parquet(fn, engine=engine)
-            df2 = dd.read_parquet(fn, engine=engine)
-            assert_eq(df, df2, check_divisions=False, check_index=False)
+        # fastparquet-442
+        warnings.simplefilter("ignore", FutureWarning)  # pandas 0.25
+        fn = str(tmpdir)
+        df = pd.DataFrame([[0]], columns=["a"], dtype="datetime64[ns, UTC]")
+        df = dd.from_pandas(df, 1)
+        df.to_parquet(fn, engine="fastparquet")
+        df2 = dd.read_parquet(fn, engine="fastparquet")
+        assert_eq(df, df2, check_divisions=False, check_index=False)
 
 
 @PYARROW_MARK
@@ -2379,21 +2318,17 @@ def test_arrow_partitioning(tmpdir):
     }
     pdf = pd.DataFrame(data)
     ddf = dd.from_pandas(pdf, npartitions=2)
-    ddf.to_parquet(path, engine="pyarrow", write_index=False, partition_on="p")
+    ddf.to_parquet(path, write_index=False, partition_on="p")
 
-    ddf = dd.read_parquet(path, index=False, engine="pyarrow")
+    ddf = dd.read_parquet(path, index=False)
 
     ddf.astype({"b": np.float32}).compute()
 
 
 def test_informative_error_messages():
-    with pytest.raises(ValueError) as info:
+    with pytest.raises(ValueError, match=r"Unsupported engine: \"foo\""):
         dd.read_parquet("foo", engine="foo")
 
-    assert "foo" in str(info.value)
-    assert "arrow" in str(info.value)
-    assert "fastparquet" in str(info.value)
-
 
 def test_append_cat_fp(tmpdir, engine):
     path = str(tmpdir)
@@ -2457,8 +2392,8 @@ def test_roundtrip_arrow(tmpdir, df):
     if not df.index.name:
         df.index.name = "index"
     ddf = dd.from_pandas(df, npartitions=2)
-    dd.to_parquet(ddf, tmp_path, engine="pyarrow", write_index=True)
-    ddf2 = dd.read_parquet(tmp_path, engine="pyarrow", calculate_divisions=True)
+    dd.to_parquet(ddf, tmp_path, write_index=True)
+    ddf2 = dd.read_parquet(tmp_path, calculate_divisions=True)
     assert_eq(ddf, ddf2)
 
 
@@ -2486,6 +2421,7 @@ def test_pathlib_path(tmpdir, engine):
 
 
 @FASTPARQUET_MARK
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 def test_categories_large(tmpdir, engine):
     # Issue #5112
     fn = str(tmpdir.join("parquet_int16.parq"))
@@ -2499,7 +2435,6 @@ def test_categories_large(tmpdir, engine):
     assert_eq(sorted(df.name.cat.categories), sorted(ddf.compute().name.cat.categories))
 
 
-@write_read_engines()
 def test_read_glob_no_meta(tmpdir, write_engine, read_engine):
     tmp_path = str(tmpdir)
     ddf.to_parquet(tmp_path, engine=write_engine)
@@ -2512,7 +2447,6 @@ def test_read_glob_no_meta(tmpdir, write_engine, read_engine):
     assert_eq(ddf, ddf2, check_divisions=False)
 
 
-@write_read_engines()
 def test_read_glob_yes_meta(tmpdir, write_engine, read_engine):
     tmp_path = str(tmpdir)
     ddf.to_parquet(tmp_path, engine=write_engine, write_metadata_file=True)
@@ -2524,7 +2458,6 @@ def test_read_glob_yes_meta(tmpdir, write_engine, read_engine):
 
 @pytest.mark.parametrize("divisions", [True, False])
 @pytest.mark.parametrize("remove_common", [True, False])
-@write_read_engines()
 def test_read_dir_nometa(tmpdir, write_engine, read_engine, divisions, remove_common):
     tmp_path = str(tmpdir)
     ddf.to_parquet(tmp_path, engine=write_engine, write_metadata_file=True)
@@ -2540,7 +2473,6 @@ def test_read_dir_nometa(tmpdir, write_engine, read_engine, divisions, remove_co
     assert_eq(ddf, ddf2, check_divisions=divisions)
 
 
-@write_read_engines()
 def test_statistics_nometa(tmpdir, write_engine, read_engine):
     tmp_path = str(tmpdir)
     ddf.to_parquet(tmp_path, engine=write_engine, write_metadata_file=False)
@@ -2921,13 +2853,10 @@ def test_filter_isna(tmpdir, split_row_groups):
         # See: https://issues.apache.org/jira/browse/ARROW-15312
         pytest.skip("pyarrow>=8.0.0 needed for correct null filtering")
     path = tmpdir.join("test.parquet")
-    pd.DataFrame({"a": [1, None] * 5 + [None] * 5}).to_parquet(
-        path, engine="pyarrow", row_group_size=10
-    )
+    pd.DataFrame({"a": [1, None] * 5 + [None] * 5}).to_parquet(path, row_group_size=10)
 
     result_isna = dd.read_parquet(
         path,
-        engine="pyarrow",
         filters=[("a", "is", np.nan)],
         split_row_groups=split_row_groups,
     )
@@ -2936,7 +2865,6 @@ def test_filter_isna(tmpdir, split_row_groups):
 
     result_notna = dd.read_parquet(
         path,
-        engine="pyarrow",
         filters=[("a", "is not", np.nan)],
         split_row_groups=split_row_groups,
     )
@@ -3010,7 +2938,6 @@ def test_optimize_and_not(tmpdir, engine):
         assert_eq(a, b)
 
 
-@write_read_engines()
 def test_split_adaptive_empty(tmpdir, write_engine, read_engine):
     df = pd.DataFrame({"a": pd.Series(dtype="int"), "b": pd.Series(dtype="float")})
     ddf1 = dd.from_pandas(df, npartitions=1)
@@ -3027,13 +2954,9 @@ def test_split_adaptive_empty(tmpdir, write_engine, read_engine):
 @pytest.mark.parametrize("metadata", [True, False])
 @pytest.mark.parametrize("partition_on", [None, "a"])
 @pytest.mark.parametrize("blocksize", [4096, "1MiB"])
-@write_read_engines()
 def test_split_adaptive_files(
     tmpdir, blocksize, partition_on, write_engine, read_engine, metadata
 ):
-    if partition_on and read_engine == "fastparquet" and not metadata:
-        pytest.skip("Fastparquet requires _metadata for partitioned data.")
-
     df_size = 100
     df1 = pd.DataFrame(
         {
@@ -3089,7 +3012,6 @@ def test_split_adaptive_files(
         assert_eq(ddf1, ddf2, check_divisions=False, check_index=False)
 
 
-@write_read_engines()
 @pytest.mark.parametrize("aggregate_files", ["a", "b"])
 def test_split_adaptive_aggregate_files(
     tmpdir, write_engine, read_engine, aggregate_files
@@ -3258,14 +3180,11 @@ def test_blocksize(tmpdir, blocksize, engine, metadata):
         assert ddf2.npartitions == ddf1.npartitions
 
 
-@write_read_engines()
 def test_roundtrip_pandas_blocksize(tmpdir, write_engine, read_engine):
     path = str(tmpdir.join("test.parquet"))
     pdf = df.copy()
     pdf.index.name = "index"
-    pdf.to_parquet(
-        path, engine="pyarrow" if write_engine.startswith("pyarrow") else "fastparquet"
-    )
+    pdf.to_parquet(path, engine=write_engine)
 
     ddf_read = dd.read_parquet(
         path,
@@ -3280,6 +3199,7 @@ def test_roundtrip_pandas_blocksize(tmpdir, write_engine, read_engine):
 
 
 @FASTPARQUET_MARK
+@pytest.mark.filterwarnings("ignore::FutureWarning")
 def test_read_pandas_fastparquet_partitioned(tmpdir, engine):
     pdf = pd.DataFrame(
         [{"str": str(i), "int": i, "group": "ABC"[i % 3]} for i in range(6)]
@@ -3296,9 +3216,7 @@ def test_read_parquet_getitem_skip_when_getting_read_parquet(tmpdir, engine):
     # https://github.com/dask/dask/issues/5893
     pdf = pd.DataFrame({"A": [1, 2, 3, 4, 5, 6], "B": ["a", "b", "c", "d", "e", "f"]})
     path = os.path.join(str(tmpdir), "data.parquet")
-    pd_engine = "pyarrow" if engine.startswith("pyarrow") else "fastparquet"
-    pdf.to_parquet(path, engine=pd_engine)
-
+    pdf.to_parquet(path, engine=engine)
     ddf = dd.read_parquet(path, engine=engine)
     a, b = dask.optimize(ddf["A"], ddf)
 
@@ -3313,7 +3231,6 @@ def test_read_parquet_getitem_skip_when_getting_read_parquet(tmpdir, engine):
 
 
 @pytest.mark.parametrize("calculate_divisions", [None, True])
-@write_read_engines()
 def test_filter_nonpartition_columns(
     tmpdir, write_engine, read_engine, calculate_divisions
 ):
@@ -3354,8 +3271,8 @@ def test_pandas_metadata_nullable_pyarrow(tmpdir):
         ),
         npartitions=1,
     )
-    ddf1.to_parquet(tmpdir, engine="pyarrow")
-    ddf2 = dd.read_parquet(tmpdir, engine="pyarrow", calculate_divisions=True)
+    ddf1.to_parquet(tmpdir)
+    ddf2 = dd.read_parquet(tmpdir, calculate_divisions=True)
 
     assert_eq(ddf1, ddf2, check_index=False)
 
@@ -3383,14 +3300,14 @@ def test_pandas_timestamp_overflow_pyarrow(tmpdir):
     if pyarrow_version < parse_version("13.0.0.dev"):
         # This will raise by default due to overflow
         with pytest.raises(pa.lib.ArrowInvalid) as e:
-            dd.read_parquet(str(tmpdir), engine="pyarrow").compute()
+            dd.read_parquet(str(tmpdir)).compute()
             assert "out of bounds" in str(e.value)
     else:
-        dd.read_parquet(str(tmpdir), engine="pyarrow").compute()
+        dd.read_parquet(str(tmpdir)).compute()
 
-    from dask.dataframe.io.parquet.arrow import ArrowDatasetEngine as ArrowEngine
+    from dask.dataframe.io.parquet.arrow import ArrowDatasetEngine
 
-    class ArrowEngineWithTimestampClamp(ArrowEngine):
+    class ArrowEngineWithTimestampClamp(ArrowDatasetEngine):
         @classmethod
         def clamp_arrow_datetimes(cls, arrow_table: pa.Table) -> pa.Table:
             """Constrain datetimes to be valid for pandas
@@ -3549,21 +3466,19 @@ def test_pyarrow_directory_partitioning(tmpdir):
     path2 = os.path.join(path2, "data.parquet")
     _df1 = pd.DataFrame({"part": "a", "col": range(5)})
     _df2 = pd.DataFrame({"part": "b", "col": range(5)})
-    _df1.to_parquet(path1, engine="pyarrow")
-    _df2.to_parquet(path2, engine="pyarrow")
+    _df1.to_parquet(path1)
+    _df2.to_parquet(path2)
 
     # Check results
     expect = pd.concat([_df1, _df2], ignore_index=True)
     result = dd.read_parquet(
         str(tmpdir),
-        engine="pyarrow",
         dataset={"partitioning": ["part"], "partition_base_dir": str(tmpdir)},
     )
     result["part"] = result["part"].astype("object")
     assert_eq(result[list(expect.columns)], expect, check_index=False)
 
 
-@fp_pandas_xfail
 def test_partitioned_preserve_index(tmpdir, write_engine, read_engine):
     tmp = str(tmpdir)
     size = 1_000
@@ -3592,50 +3507,32 @@ def test_partitioned_preserve_index(tmpdir, write_engine, read_engine):
 
 
 def test_from_pandas_preserve_none_index(tmpdir, engine):
-    if engine.startswith("pyarrow"):
-        pytest.importorskip("pyarrow", minversion="0.15.0")
-
     fn = str(tmpdir.join("test.parquet"))
     df = pd.DataFrame({"a": [1, 2], "b": [4, 5], "c": [6, 7]}).set_index("c")
     df.index.name = None
-    df.to_parquet(
-        fn,
-        engine="pyarrow" if engine.startswith("pyarrow") else "fastparquet",
-        index=True,
-    )
-
+    df.to_parquet(fn, engine=engine, index=True)
     expect = pd.read_parquet(fn)
     got = dd.read_parquet(fn, engine=engine)
     assert_eq(expect, got)
 
 
 def test_multi_partition_none_index_false(tmpdir, engine):
-    if engine.startswith("pyarrow"):
-        pytest.importorskip("pyarrow", minversion="0.15.0")
-        write_engine = "pyarrow"
-    else:
-        assert engine == "fastparquet"
-        write_engine = "fastparquet"
-
     # Write dataset without dask.to_parquet
     ddf1 = ddf.reset_index(drop=True)
     for i, part in enumerate(ddf1.partitions):
         path = tmpdir.join(f"test.{i}.parquet")
-        part.compute().to_parquet(str(path), engine=write_engine)
+        part.compute().to_parquet(str(path), engine=engine)
 
     # Read back with index=False
     ddf2 = dd.read_parquet(str(tmpdir), index=False, engine=engine)
     assert_eq(ddf1, ddf2)
 
 
-@write_read_engines()
 def test_from_pandas_preserve_none_rangeindex(tmpdir, write_engine, read_engine):
     # See GitHub Issue#6348
     fn = str(tmpdir.join("test.parquet"))
     df0 = pd.DataFrame({"t": [1, 2, 3]}, index=pd.RangeIndex(start=1, stop=4))
-    df0.to_parquet(
-        fn, engine="pyarrow" if write_engine.startswith("pyarrow") else "fastparquet"
-    )
+    df0.to_parquet(fn, engine=write_engine)
 
     df1 = dd.read_parquet(fn, engine=read_engine)
     assert_eq(df0, df1.compute())
@@ -3707,7 +3604,6 @@ def test_pyarrow_dataset_partitioned(tmpdir, engine, test_filter):
 @PYARROW_MARK
 @pytest.mark.parametrize("scheduler", [None, "processes"])
 def test_null_partition_pyarrow(tmpdir, scheduler):
-    engine = "pyarrow"
     df = pd.DataFrame(
         {
             "id": pd.Series([0, 1, None], dtype="Int64"),
@@ -3715,14 +3611,13 @@ def test_null_partition_pyarrow(tmpdir, scheduler):
         }
     )
     ddf = dd.from_pandas(df, npartitions=1)
-    ddf.to_parquet(str(tmpdir), engine=engine, partition_on="id")
+    ddf.to_parquet(str(tmpdir), partition_on="id")
     fns = glob.glob(os.path.join(tmpdir, "id=*/*.parquet"))
     assert len(fns) == 3
 
     # Check proper partitioning usage
     ddf_read = dd.read_parquet(
         str(tmpdir),
-        engine=engine,
         dtype_backend="numpy_nullable",
         dataset={
             "partitioning": {
@@ -3751,21 +3646,14 @@ def test_pyarrow_dataset_read_from_paths(tmpdir):
     df = pd.DataFrame({"a": [4, 5, 6], "b": ["a", "b", "b"]})
     df["b"] = df["b"].astype("category")
     ddf = dd.from_pandas(df, npartitions=2)
-    ddf.to_parquet(fn, engine="pyarrow", partition_on="b")
+    ddf.to_parquet(fn, partition_on="b")
 
     with pytest.warns(FutureWarning):
         read_df_1 = dd.read_parquet(
-            fn,
-            engine="pyarrow",
-            filters=[("b", "==", "a")],
-            read_from_paths=False,
+            fn, filters=[("b", "==", "a")], read_from_paths=False
         )
 
-    read_df_2 = dd.read_parquet(
-        fn,
-        engine="pyarrow",
-        filters=[("b", "==", "a")],
-    )
+    read_df_2 = dd.read_parquet(fn, filters=[("b", "==", "a")])
 
     assert_eq(read_df_1, read_df_2)
     assert_eq(ddf[ddf["b"] == "a"].compute(), read_df_2.compute())
@@ -3784,12 +3672,11 @@ def test_pyarrow_dataset_filter_partitioned(tmpdir, split_row_groups):
     )
     df["b"] = df["b"].astype("category")
     ddf = dd.from_pandas(df, npartitions=2)
-    ddf.to_parquet(fn, engine="pyarrow", partition_on=["b", "c"])
+    ddf.to_parquet(fn, partition_on=["b", "c"])
 
     # Filter on a a non-partition column
     read_df = dd.read_parquet(
         fn,
-        engine="pyarrow",
         split_row_groups=split_row_groups,
         filters=[("a", "==", 5)],
     )
@@ -3818,7 +3705,7 @@ def test_pyarrow_dataset_filter_on_partitioned(tmpdir, engine):
     read_ddf["part"] = read_ddf["part"].astype("object")
     assert_eq(df.iloc[2:3], read_ddf)
 
-    # Check that List[List[Tuple]] filters are aplied.
+    # Check that List[List[Tuple]] filters are applied.
     # (fastparquet doesn't support this format)
     if engine == "pyarrow":
         read_ddf = dd.read_parquet(
@@ -3848,7 +3735,6 @@ def test_parquet_pyarrow_write_empty_metadata(tmpdir):
     df = dd.from_delayed([df_a, df_b, df_c])
     df.to_parquet(
         tmpdir,
-        engine="pyarrow",
         partition_on=["x"],
         append=False,
         write_metadata_file=True,
@@ -3883,7 +3769,6 @@ def test_parquet_pyarrow_write_empty_metadata_append(tmpdir):
     df1 = dd.from_delayed([df_a, df_b])
     df1.to_parquet(
         tmpdir,
-        engine="pyarrow",
         partition_on=["x"],
         append=False,
         write_metadata_file=True,
@@ -3899,7 +3784,6 @@ def test_parquet_pyarrow_write_empty_metadata_append(tmpdir):
     df2 = dd.from_delayed([df_c, df_d])
     df2.to_parquet(
         tmpdir,
-        engine="pyarrow",
         partition_on=["x"],
         append=True,
         ignore_divisions=True,
@@ -3909,7 +3793,6 @@ def test_parquet_pyarrow_write_empty_metadata_append(tmpdir):
 
 @PYARROW_MARK
 @pytest.mark.parametrize("partition_on", [None, "a"])
-@write_read_engines()
 def test_create_metadata_file(tmpdir, write_engine, read_engine, partition_on):
     tmpdir = str(tmpdir)
 
@@ -4139,8 +4022,8 @@ def test_roundtrip_decimal_dtype(tmpdir):
     ]
     ddf1 = dd.from_pandas(pd.DataFrame(data), npartitions=1)
 
-    ddf1.to_parquet(path=tmpdir, engine="pyarrow", schema={"col1": pa.decimal128(5, 2)})
-    ddf2 = dd.read_parquet(tmpdir, engine="pyarrow")
+    ddf1.to_parquet(path=tmpdir, schema={"col1": pa.decimal128(5, 2)})
+    ddf2 = dd.read_parquet(tmpdir)
 
     if pyarrow_strings_enabled():
         assert pa.types.is_decimal(ddf2["col1"].dtype.pyarrow_dtype)
@@ -4164,8 +4047,8 @@ def test_roundtrip_date_dtype(tmpdir):
     ]
     ddf1 = dd.from_pandas(pd.DataFrame(data), npartitions=1)
 
-    ddf1.to_parquet(path=tmpdir, engine="pyarrow", schema={"col1": pa.date32()})
-    ddf2 = dd.read_parquet(tmpdir, engine="pyarrow")
+    ddf1.to_parquet(path=tmpdir, schema={"col1": pa.date32()})
+    ddf2 = dd.read_parquet(tmpdir)
 
     if pyarrow_strings_enabled():
         assert pa.types.is_date32(ddf2["col1"].dtype.pyarrow_dtype)
@@ -4426,7 +4309,8 @@ def test_custom_filename(tmpdir, engine):
     assert_eq(df, dd.read_parquet(fn, engine=engine, calculate_divisions=True))
 
 
-def test_custom_filename_works_with_pyarrow_when_append_is_true(tmpdir, engine):
+@PYARROW_MARK
+def test_custom_filename_works_with_pyarrow_when_append_is_true(tmpdir):
     fn = str(tmpdir)
     pdf = pd.DataFrame(
         {"num1": [1, 2, 3, 4], "num2": [7, 8, 9, 10]},
@@ -4436,23 +4320,15 @@ def test_custom_filename_works_with_pyarrow_when_append_is_true(tmpdir, engine):
         fn,
         write_metadata_file=True,
         name_function=lambda x: f"hi-{x * 2}.parquet",
-        engine=engine,
     )
 
     pdf = pd.DataFrame(
         {"num1": [33], "num2": [44]},
     )
     df = dd.from_pandas(pdf, npartitions=1)
-    if engine == "fastparquet":
-        pytest.xfail(
-            "fastparquet errors our with IndexError when ``name_function`` is customized "
-            "and append is set to True.  We didn't do a detailed investigation for expediency. "
-            "See this comment for the conversation: https://github.com/dask/dask/pull/7682#issuecomment-845243623"
-        )
     df.to_parquet(
         fn,
         name_function=lambda x: f"hi-{x * 2}.parquet",
-        engine=engine,
         append=True,
         ignore_divisions=True,
     )
@@ -4465,7 +4341,7 @@ def test_custom_filename_works_with_pyarrow_when_append_is_true(tmpdir, engine):
     expected_pdf = pd.DataFrame(
         {"num1": [1, 2, 3, 4, 33], "num2": [7, 8, 9, 10, 44]},
     )
-    actual = dd.read_parquet(fn, engine=engine, index=False)
+    actual = dd.read_parquet(fn, index=False)
     assert_eq(actual, expected_pdf, check_index=False)
 
 
@@ -4653,11 +4529,7 @@ def test_deprecate_gather_statistics(tmp_path, engine):
     path = tmp_path / "test_deprecate_gather_statistics.parquet"
     df.to_parquet(path, engine=engine)
     with pytest.warns(FutureWarning, match="deprecated"):
-        out = dd.read_parquet(
-            path,
-            engine=engine,
-            gather_statistics=True,
-        )
+        out = dd.read_parquet(path, engine=engine, gather_statistics=True)
     assert_eq(out, df)
 
 
@@ -4737,11 +4609,7 @@ def test_filesystem_option(tmp_path, engine, fs):
     df = pd.DataFrame({"a": range(10)})
     dd.from_pandas(df, npartitions=2).to_parquet(tmp_path, engine=engine)
     filesystem = fs or LocalFileSystem()
-    ddf = dd.read_parquet(
-        tmp_path,
-        engine=engine,
-        filesystem=filesystem,
-    )
+    ddf = dd.read_parquet(tmp_path, engine=engine, filesystem=filesystem)
     if fs is None:
         layer_fs = next(iter(ddf.dask.layers.values())).io_func.fs
         assert layer_fs is filesystem
@@ -4757,11 +4625,7 @@ def test_pyarrow_filesystem_option(tmp_path, fs):
     df = pd.DataFrame({"a": range(10)})
     fs = fs or LocalFileSystem()
     dd.from_pandas(df, npartitions=2).to_parquet(tmp_path, filesystem=fs)
-    ddf = dd.read_parquet(
-        tmp_path,
-        engine="pyarrow",
-        filesystem=fs,
-    )
+    ddf = dd.read_parquet(tmp_path, filesystem=fs)
     layer_fs = next(iter(ddf.dask.layers.values())).io_func.fs
     assert isinstance(layer_fs, ArrowFSWrapper)
     assert isinstance(layer_fs.fs, LocalFileSystem)
@@ -4790,16 +4654,12 @@ def test_fsspec_to_parquet_filesystem_option(tmp_path):
 
     df = pd.DataFrame({"a": range(10)})
     fs = get_filesystem_class("memory")(use_instance_cache=False)
-    df.to_parquet(key1, engine="pyarrow", filesystem=fs)
+    df.to_parquet(key1, filesystem=fs)
 
     # read in prepared data
-    ddf = dd.read_parquet(
-        key1,
-        engine="pyarrow",
-        filesystem=fs,
-    )
+    ddf = dd.read_parquet(key1, filesystem=fs)
     assert_eq(ddf, df)
-    ddf.to_parquet(key2, engine="pyarrow", filesystem=fs)
+    ddf.to_parquet(key2, filesystem=fs)
 
     # make sure we didn't write to local fs
     assert len(list(tmp_path.iterdir())) == 0, "wrote to local fs"
@@ -4808,18 +4668,21 @@ def test_fsspec_to_parquet_filesystem_option(tmp_path):
     assert len(fs.ls(key2, detail=False)) == 1
 
     # ensure append functionality works
-    ddf.to_parquet(key2, engine="pyarrow", append=True, filesystem=fs)
+    ddf.to_parquet(key2, append=True, filesystem=fs)
     assert len(fs.ls(key2, detail=False)) == 2, "should have two parts"
 
-    rddf = dd.read_parquet(key2, engine="pyarrow", filesystem=fs)
+    rddf = dd.read_parquet(key2, filesystem=fs)
     assert_eq(rddf, dd.concat([ddf, ddf]))
 
 
 def test_select_filtered_column(tmp_path, engine):
     df = pd.DataFrame({"a": range(10), "b": ["cat"] * 10})
     path = tmp_path / "test_select_filtered_column.parquet"
-    stats = {"write_statistics" if engine == "pyarrow" else "stats": True}
-    df.to_parquet(path, engine=engine, index=False, **stats)
+
+    if engine == "fastparquet":
+        df.to_parquet(path, index=False, stats=True, engine="fastparquet")
+    else:
+        df.to_parquet(path, index=False, write_statistics=True)
 
     with pytest.warns(UserWarning, match="Sorted columns detected"):
         ddf = dd.read_parquet(path, engine=engine, filters=[("b", "==", "cat")])
@@ -4833,8 +4696,11 @@ def test_select_filtered_column(tmp_path, engine):
 def test_select_filtered_column_no_stats(tmp_path, engine):
     df = pd.DataFrame({"a": range(10), "b": ["cat"] * 10})
     path = tmp_path / "test_select_filtered_column_no_stats.parquet"
-    stats = {"write_statistics" if engine == "pyarrow" else "stats": False}
-    df.to_parquet(path, engine=engine, **stats)
+
+    if engine == "fastparquet":
+        df.to_parquet(path, stats=False, engine="fastparquet")
+    else:
+        df.to_parquet(path, write_statistics=False)
 
     ddf = dd.read_parquet(path, engine=engine, filters=[("b", "==", "cat")])
     assert_eq(df, ddf)
@@ -4843,6 +4709,7 @@ def test_select_filtered_column_no_stats(tmp_path, engine):
     assert_eq(df, ddf)
 
 
+@PYARROW_MARK
 @pytest.mark.parametrize("convert_string", [True, False])
 @pytest.mark.skipif(
     not PANDAS_GE_200, reason="dataframe.convert-string requires pandas>=2.0"
@@ -4969,8 +4836,8 @@ def test_read_parquet_preserve_categorical_column_dtype(tmp_path):
     df = pd.DataFrame({"a": [1, 2], "b": ["x", "y"]})
 
     outdir = tmp_path / "out.parquet"
-    df.to_parquet(outdir, engine="pyarrow", partition_cols=["a"])
-    ddf = dd.read_parquet(outdir, engine="pyarrow")
+    df.to_parquet(outdir, partition_cols=["a"])
+    ddf = dd.read_parquet(outdir)
 
     expected = pd.DataFrame(
         {"b": ["x", "y"], "a": pd.Categorical(pd.Index([1, 2], dtype="int32"))},
@@ -4984,9 +4851,9 @@ def test_read_parquet_preserve_categorical_column_dtype(tmp_path):
 def test_dtype_backend_categoricals(tmp_path):
     df = pd.DataFrame({"a": pd.Series(["x", "y"], dtype="category"), "b": [1, 2]})
     outdir = tmp_path / "out.parquet"
-    df.to_parquet(outdir, engine="pyarrow")
-    ddf = dd.read_parquet(outdir, engine="pyarrow", dtype_backend="pyarrow")
-    pdf = pd.read_parquet(outdir, engine="pyarrow", dtype_backend="pyarrow")
+    df.to_parquet(outdir)
+    ddf = dd.read_parquet(outdir, dtype_backend="pyarrow")
+    pdf = pd.read_parquet(outdir, dtype_backend="pyarrow")
     # Set sort_results=False because of pandas bug up to 2.0.1
     assert_eq(ddf, pdf, sort_results=PANDAS_GE_202)
 
@@ -4998,23 +4865,15 @@ def test_non_categorical_partitioning_pyarrow(tmpdir, filters):
 
     df1 = pd.DataFrame({"a": range(100), "b": ["cat", "dog"] * 50})
     ddf1 = dd.from_pandas(df1, npartitions=2)
-    ddf1.to_parquet(
-        path=tmpdir, partition_on=["b"], write_index=False, engine="pyarrow"
-    )
+    ddf1.to_parquet(path=tmpdir, partition_on=["b"], write_index=False)
 
     schema = pa.schema([("b", pa.string())])
     partitioning = dict(flavor="hive", schema=schema)
     ddf = dd.read_parquet(
-        tmpdir,
-        dataset={"partitioning": partitioning},
-        filters=filters,
-        engine="pyarrow",
+        tmpdir, dataset={"partitioning": partitioning}, filters=filters
     )
     pdf = pd.read_parquet(
-        tmpdir,
-        partitioning=pd_partitioning(**partitioning),
-        filters=filters,
-        engine="pyarrow",
+        tmpdir, partitioning=pd_partitioning(**partitioning), filters=filters
     )
     assert_eq(ddf, pdf, check_index=False)
     assert ddf["b"].dtype != "category"
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 9c27b0280..495251608 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -5744,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():
@@ -6007,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_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_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..de48d3ab4 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"),
@@ -482,15 +482,15 @@ def test_blockwise_dataframe_io(c, tmpdir, io, fuse, from_futures):
     else:
         ddf0 = dd.from_pandas(df, npartitions=3)
 
-    if io.startswith("parquet"):
-        if io == "parquet-pyarrow":
-            pytest.importorskip("pyarrow.parquet")
-            engine = "pyarrow"
-        else:
-            pytest.importorskip("fastparquet")
-            engine = "fastparquet"
-        ddf0.to_parquet(str(tmpdir), engine=engine)
-        ddf = dd.read_parquet(str(tmpdir), engine=engine)
+    if io == "parquet-pyarrow":
+        pytest.importorskip("pyarrow")
+        ddf0.to_parquet(str(tmpdir))
+        ddf = dd.read_parquet(str(tmpdir))
+    elif io == "parquet-fastparquet":
+        pytest.importorskip("fastparquet")
+        with pytest.warns(FutureWarning):
+            ddf0.to_parquet(str(tmpdir), engine="fastparquet")
+            ddf = dd.read_parquet(str(tmpdir), engine="fastparquet")
     elif io == "csv":
         ddf0.to_csv(str(tmpdir), index=False)
         ddf = dd.read_csv(os.path.join(str(tmpdir), "*"))
@@ -499,6 +499,8 @@ def test_blockwise_dataframe_io(c, tmpdir, io, fuse, from_futures):
         fn = str(tmpdir.join("h5"))
         ddf0.to_hdf(fn, "/data*")
         ddf = dd.read_hdf(fn, "/data*")
+    else:
+        raise AssertionError("unreachable")
 
     df = df[["x"]] + 10
     ddf = ddf[["x"]] + 10
@@ -613,7 +615,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 +721,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 +767,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..3d3e28df8 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):
@@ -134,34 +134,17 @@ def test_fractional_slice():
 
 
 def _pq_pyarrow(tmpdir):
-    pytest.importorskip("pyarrow.parquet")
+    pytest.importorskip("pyarrow")
     pd = pytest.importorskip("pandas")
     dd = pytest.importorskip("dask.dataframe")
 
-    try:
-        import pyarrow.dataset as pa_ds
-    except ImportError:
-        # PyArrow version too old for Dataset API
-        pa_ds = None
-
     dd.from_pandas(pd.DataFrame({"a": range(10)}), npartitions=2).to_parquet(
-        str(tmpdir),
-        engine="pyarrow",
+        str(tmpdir)
     )
     filters = [(("a", "<=", 2))]
 
-    ddf1 = dd.read_parquet(str(tmpdir), engine="pyarrow", filters=filters)
-    if pa_ds:
-        # Need to test that layer serialization succeeds
-        # with "pyarrow-dataset" filtering
-        ddf2 = dd.read_parquet(
-            str(tmpdir),
-            engine="pyarrow-dataset",
-            filters=filters,
-        )
-        return (ddf1, ddf2)
-    else:
-        return ddf1
+    ddf1 = dd.read_parquet(str(tmpdir), filters=filters)
+    return ddf1
 
 
 def _pq_fastparquet(tmpdir):
@@ -169,11 +152,10 @@ def _pq_fastparquet(tmpdir):
     pd = pytest.importorskip("pandas")
     dd = pytest.importorskip("dask.dataframe")
 
-    dd.from_pandas(pd.DataFrame({"a": range(10)}), npartitions=2).to_parquet(
-        str(tmpdir),
-        engine="fastparquet",
-    )
-    return dd.read_parquet(str(tmpdir), engine="fastparquet")
+    df = dd.from_pandas(pd.DataFrame({"a": range(10)}), npartitions=2)
+    with pytest.warns(FutureWarning):
+        df.to_parquet(str(tmpdir), engine="fastparquet")
+        return dd.read_parquet(str(tmpdir), engine="fastparquet")
 
 
 def _read_csv(tmpdir):
@@ -232,7 +214,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):
diff --git a/dask/tests/test_spark_compat.py b/dask/tests/test_spark_compat.py
index a4a469b2a..473472ce5 100644
--- a/dask/tests/test_spark_compat.py
+++ b/dask/tests/test_spark_compat.py
@@ -33,6 +33,20 @@ pytestmark = [
     pytest.mark.skip_with_pyarrow_strings,  # pyspark doesn't support pandas 2.0
 ]
 
+
+@pytest.fixture(
+    params=[
+        "pyarrow",
+        pytest.param(
+            "fastparquet", marks=pytest.mark.filterwarnings("ignore::FutureWarning")
+        ),
+    ]
+)
+def engine(request):
+    pytest.importorskip(request.param)
+    return request.param
+
+
 # pyspark auto-converts timezones -- round-tripping timestamps is easier if
 # we set everything to UTC.
 pdf = timeseries(freq="1h").compute()
@@ -62,8 +76,7 @@ def spark_session():
         signal.signal(signal.SIGINT, prev)
 
 
-@pytest.mark.parametrize("npartitions", (1, 5, 10))
-@pytest.mark.parametrize("engine", ("pyarrow", "fastparquet"))
+@pytest.mark.parametrize("npartitions", [1, 5, 10])
 def test_roundtrip_parquet_spark_to_dask(spark_session, npartitions, tmpdir, engine):
     tmpdir = str(tmpdir)
 
@@ -80,7 +93,6 @@ def test_roundtrip_parquet_spark_to_dask(spark_session, npartitions, tmpdir, eng
     assert_eq(ddf, pdf, check_index=False)
 
 
-@pytest.mark.parametrize("engine", ("pyarrow", "fastparquet"))
 def test_roundtrip_hive_parquet_spark_to_dask(spark_session, tmpdir, engine):
     tmpdir = str(tmpdir)
 
@@ -103,8 +115,7 @@ def test_roundtrip_hive_parquet_spark_to_dask(spark_session, tmpdir, engine):
     assert_eq(ddf, pdf.sort_index(axis=1), check_index=False)
 
 
-@pytest.mark.parametrize("npartitions", (1, 5, 10))
-@pytest.mark.parametrize("engine", ("pyarrow", "fastparquet"))
+@pytest.mark.parametrize("npartitions", [1, 5, 10])
 def test_roundtrip_parquet_dask_to_spark(spark_session, npartitions, tmpdir, engine):
     tmpdir = str(tmpdir)
     ddf = dd.from_pandas(pdf, npartitions=npartitions)