Repository: dask/dask

Issue:
dask.dataframe.read_parquet producing empty columns for those partitioned on
When using pyarrow as the engine in read_parquet, I'm encountering an error working with the columns on which I've partitioned.

If I select the column that I partitioned on, I get an empty series back.

I am using dask version ('2.8.1'), pyarrow ('0.15.1'), nd fastparquet ('0.3.2').
```
$ conda create -c conda-forge -n dask.arrow.issue -y dask fastparquet pyarrow jupyterlab python-snappy
```

To reproduce

```
import dask, pyarrow, fastparquet
(dask.__version__, pyarrow.__version__, fastparquet.__version__) # ('2.9.1', '0.15.1', '0.3.2')

import pandas as pd
import dask.dataframe as dd
import pyarrow.parquet as pq

df = pd.DataFrame([{'str': str(i), 'int':i, 'group': 'ABC'[i%3]} for i in range(6)])

pq_dir = './parquet_partition_issue'
df.to_parquet(pq_dir, partition_cols=['group'], engine='fastparquet')

ddf_arrow = dd.read_parquet(pq_dir, engine='pyarrow')
ddf_fast = dd.read_parquet(pq_dir, engine='fastparquet')

print(len(ddf_arrow['group'].compute()))  # 0
print(len(ddf_fast['group'].compute()))  # 6
print(len(ddf_arrow.compute().group))  # 6
print(len(pq.read_table(pq_dir).to_pandas().group))  # 6
```


Discussion:

Patch:
diff --git a/dask/dataframe/_compat.py b/dask/dataframe/_compat.py
new file mode 100644
index 000000000..16453b0ad
--- /dev/null
+++ b/dask/dataframe/_compat.py
@@ -0,0 +1,85 @@
+import string
+from distutils.version import LooseVersion
+
+import numpy as np
+import pandas as pd
+
+
+PANDAS_VERSION = LooseVersion(pd.__version__)
+PANDAS_GT_0230 = PANDAS_VERSION >= LooseVersion("0.23.0")
+PANDAS_GT_0240 = PANDAS_VERSION >= LooseVersion("0.24.0")
+PANDAS_GT_0250 = PANDAS_VERSION >= LooseVersion("0.25.0")
+PANDAS_GT_100 = PANDAS_VERSION >= LooseVersion("1.0.0rc0")
+HAS_INT_NA = PANDAS_GT_0240
+
+
+if PANDAS_GT_100:
+    import pandas.testing as tm  # noqa: F401
+else:
+    import pandas.util.testing as tm  # noqa: F401
+
+
+def assert_categorical_equal(left, right, *args, **kwargs):
+    if PANDAS_GT_100:
+        tm.assert_extension_array_equal(left, right, *args, **kwargs)
+        assert pd.api.types.is_categorical_dtype(
+            left.dtype
+        ), "{} is not categorical dtype".format(left)
+        assert pd.api.types.is_categorical_dtype(
+            right.dtype
+        ), "{} is not categorical dtype".format(right)
+    else:
+        return tm.assert_categorical_equal(left, right, *args, **kwargs)
+
+
+def assert_numpy_array_equal(left, right):
+    left_na = pd.isna(left)
+    right_na = pd.isna(right)
+    np.testing.assert_array_equal(left_na, right_na)
+
+    left_valid = left[~left_na]
+    right_valid = right[~right_na]
+    np.testing.assert_array_equal(left_valid, right_valid)
+
+
+def makeDataFrame():
+    data = np.random.randn(30, 4)
+    index = list(string.ascii_letters)[:30]
+    return pd.DataFrame(data, index=index, columns=list("ABCD"))
+
+
+def makeTimeDataFrame():
+    data = makeDataFrame()
+    data.index = makeDateIndex()
+    return data
+
+
+def makeTimeSeries():
+    return makeTimeDataFrame()["A"]
+
+
+def makeDateIndex(k=30, freq="B"):
+    return pd.date_range("2000", periods=k, freq=freq)
+
+
+def makeTimedeltaIndex(k=30, freq="D"):
+    return pd.timedelta_range("1 day", periods=k, freq=freq)
+
+
+def makeMissingDataframe():
+    df = makeDataFrame()
+    data = df.values
+    data = np.where(data > 1, np.nan, data)
+    return pd.DataFrame(data, index=df.index, columns=df.columns)
+
+
+def makeMixedDataFrame():
+    df = pd.DataFrame(
+        {
+            "A": [0.0, 1, 2, 3, 4],
+            "B": [0.0, 1, 0, 1, 0],
+            "C": ["foo{}".format(i) for i in range(5)],
+            "D": pd.date_range("2009-01-01", periods=5),
+        }
+    )
+    return df
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 58547409d..cccd4307a 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1709,9 +1709,11 @@ Dask Name: {name}, {task} tasks""".format(
                 self._meta_nonempty.select_dtypes(include=[np.timedelta64]).columns
             )
 
-            if count_timedeltas == len(self._meta.columns):
+            # pandas 1.0+ does not implement var on timedelta
+
+            if not PANDAS_GT_100 and count_timedeltas == len(self._meta.columns):
                 result = self._var_timedeltas(skipna, ddof, split_every)
-            elif count_timedeltas > 0:
+            elif not PANDAS_GT_100 and count_timedeltas > 0:
                 result = self._var_mixed(skipna, ddof, split_every)
             else:
                 result = self._var_numeric(skipna, ddof, split_every)
@@ -2109,7 +2111,11 @@ Dask Name: {name}, {task} tasks""".format(
 
             name2 = "{0}{1}-take-last".format(self._token_prefix, op_name)
             cumlast = map_partitions(
-                _take_last, cumpart, skipna, meta=pd.Series([]), token=name2
+                _take_last,
+                cumpart,
+                skipna,
+                meta=pd.Series([], dtype="float"),
+                token=name2,
             )
 
             suffix = tokenize(self)
@@ -2337,7 +2343,12 @@ Dask Name: {name}, {task} tasks""".format(
         date = self.divisions[0] + offset
         end = self.loc._get_partitions(date)
 
-        include_right = offset.isAnchored() or not hasattr(offset, "_inc")
+        if PANDAS_GT_100:
+            is_anchored = offset.is_anchored()
+        else:
+            is_anchored = offset.isAnchored()
+
+        include_right = is_anchored or not hasattr(offset, "_inc")
 
         if end == self.npartitions - 1:
             divs = self.divisions
@@ -5114,7 +5125,7 @@ def quantile(df, q, method="default"):
         name = "quantiles-" + token
         empty_index = pd.Index([], dtype=float)
         return Series(
-            {(name, 0): pd.Series([], name=df.name, index=empty_index)},
+            {(name, 0): pd.Series([], name=df.name, index=empty_index, dtype="float")},
             name,
             df._meta,
             [None, None],
@@ -5387,7 +5398,7 @@ def _take_last(a, skipna=True):
             # create Series from appropriate backend dataframe library
             series_typ = type(a.iloc[0:1, 0])
             if a.empty:
-                return series_typ([])
+                return series_typ([], dtype="float")
             return series_typ(
                 {col: _last_valid(a[col]) for col in a.columns}, index=a.columns
             )
@@ -5887,13 +5898,18 @@ def maybe_shift_divisions(df, periods, freq):
     """
     if isinstance(freq, str):
         freq = pd.tseries.frequencies.to_offset(freq)
-    if isinstance(freq, pd.DateOffset) and (
-        freq.isAnchored() or not hasattr(freq, "delta")
-    ):
-        # Can't infer divisions on relative or anchored offsets, as
-        # divisions may now split identical index value.
-        # (e.g. index_partitions = [[1, 2, 3], [3, 4, 5]])
-        return df.clear_divisions()
+
+    is_offset = isinstance(freq, pd.DateOffset)
+    if is_offset:
+        if PANDAS_GT_100:
+            is_anchored = freq.is_anchored()
+        else:
+            is_anchored = freq.isAnchored()
+        if is_anchored or not hasattr(freq, "delta"):
+            # Can't infer divisions on relative or anchored offsets, as
+            # divisions may now split identical index value.
+            # (e.g. index_partitions = [[1, 2, 3], [3, 4, 5]])
+            return df.clear_divisions()
     if df.known_divisions:
         divs = pd.Series(range(len(df.divisions)), index=df.divisions)
         divisions = divs.shift(periods, freq=freq).index
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index 708a3ab7b..6ae269adc 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -342,6 +342,16 @@ class ArrowEngine(Engine):
                 open_file_func=partial(fs.open, mode="rb"),
             )
 
+        # Ensure `columns` and `partitions` do not overlap
+        columns_and_parts = columns.copy()
+        if columns_and_parts and partitions:
+            for part_name in partitions.partition_names:
+                if part_name in columns:
+                    columns.remove(part_name)
+                else:
+                    columns_and_parts.append(part_name)
+            columns = columns or None
+
         df = piece.read(
             columns=columns,
             partitions=partitions,
@@ -349,7 +359,7 @@ class ArrowEngine(Engine):
             use_threads=False,
             **kwargs.get("read", {}),
         ).to_pandas(categories=categories, use_threads=False, ignore_metadata=True)[
-            list(columns)
+            list(columns_and_parts)
         ]
 
         if index:
diff --git a/dask/dataframe/multi.py b/dask/dataframe/multi.py
index 598176e09..1f15a1c7d 100644
--- a/dask/dataframe/multi.py
+++ b/dask/dataframe/multi.py
@@ -64,6 +64,7 @@ from pandas.api.types import is_dtype_equal
 from ..base import tokenize, is_dask_collection
 from ..highlevelgraph import HighLevelGraph
 from ..utils import apply
+from ._compat import PANDAS_GT_100
 from .core import (
     _Frame,
     DataFrame,
@@ -715,7 +716,10 @@ def concat_and_unsort(frames, columns):
 
 
 def _concat_compat(frames, left, right):
-    if PANDAS_GT_0230:
+    if PANDAS_GT_100:
+        # join_axes removed
+        return (pd.concat, frames, 0, "outer", False, None, None, None, False, False)
+    elif PANDAS_GT_0230:
         # (axis, join, join_axis, ignore_index, keys, levels, names, verify_integrity, sort)
         # we only care about sort, to silence warnings.
         return (
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index b77f14582..93b0c8de8 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -3,7 +3,6 @@ import numbers
 import re
 import textwrap
 from collections.abc import Iterator, Mapping
-from distutils.version import LooseVersion
 
 import sys
 import traceback
@@ -11,7 +10,6 @@ from contextlib import contextmanager
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 from pandas.api.types import (
     is_categorical_dtype,
     is_scalar,
@@ -21,6 +19,17 @@ from pandas.api.types import (
     is_interval_dtype,
 )
 
+# include these here for compat
+from ._compat import (  # noqa: F401
+    PANDAS_VERSION,
+    PANDAS_GT_0230,
+    PANDAS_GT_0240,
+    PANDAS_GT_0250,
+    PANDAS_GT_100,
+    HAS_INT_NA,
+    tm,
+)
+
 from .extensions import make_array_nonempty, make_scalar
 from ..base import is_dask_collection
 from ..core import get_deps
@@ -31,14 +40,6 @@ from ..utils import is_series_like as dask_is_series_like
 from ..utils import is_index_like as dask_is_index_like
 
 
-PANDAS_VERSION = LooseVersion(pd.__version__)
-PANDAS_GT_0230 = PANDAS_VERSION >= LooseVersion("0.23.0")
-PANDAS_GT_0240 = PANDAS_VERSION >= LooseVersion("0.24.0rc1")
-PANDAS_GT_0250 = PANDAS_VERSION >= LooseVersion("0.25.0")
-PANDAS_GT_100 = PANDAS_VERSION >= LooseVersion("0.26.0.dev0")
-HAS_INT_NA = PANDAS_GT_0240
-
-
 def is_integer_na_dtype(t):
     dtype = getattr(t, "dtype", t)
     if HAS_INT_NA:


Test patch:
diff --git a/dask/bytes/tests/test_s3.py b/dask/bytes/tests/test_s3.py
index 871e54de9..050ba5ccb 100644
--- a/dask/bytes/tests/test_s3.py
+++ b/dask/bytes/tests/test_s3.py
@@ -386,6 +386,8 @@ def test_modification_time_read_bytes():
 @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
 def test_parquet(s3, engine):
     dd = pytest.importorskip("dask.dataframe")
+    from dask.dataframe._compat import tm
+
     lib = pytest.importorskip(engine)
     if engine == "pyarrow" and LooseVersion(lib.__version__) < "0.13.1":
         pytest.skip("pyarrow < 0.13.1 not supported for parquet")
@@ -415,7 +417,7 @@ def test_parquet(s3, engine):
     df2 = dd.read_parquet(url, index="foo", engine=engine)
     assert len(df2.divisions) > 1
 
-    pd.util.testing.assert_frame_equal(data, df2.compute())
+    tm.assert_frame_equal(data, df2.compute())
 
 
 def test_parquet_wstoragepars(s3):
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index 50ec4b7ab..5f6dbacfe 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -11,10 +11,9 @@ dd = pytest.importorskip("dask.dataframe")
 
 from toolz import partition_all, valmap
 
-import pandas.util.testing as tm
-
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.base import compute_as_if_collection
 from dask.dataframe.io.csv import (
     text_blocks_to_pandas,
diff --git a/dask/dataframe/io/tests/test_demo.py b/dask/dataframe/io/tests/test_demo.py
index c6d0d7e00..f8bbb11a7 100644
--- a/dask/dataframe/io/tests/test_demo.py
+++ b/dask/dataframe/io/tests/test_demo.py
@@ -1,9 +1,9 @@
-import pandas.util.testing as tm
 import pandas as pd
 import pytest
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.dataframe.utils import assert_eq
 
 
diff --git a/dask/dataframe/io/tests/test_hdf.py b/dask/dataframe/io/tests/test_hdf.py
index 9909c5990..949b4ed4a 100644
--- a/dask/dataframe/io/tests/test_hdf.py
+++ b/dask/dataframe/io/tests/test_hdf.py
@@ -1,6 +1,5 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import os
@@ -9,6 +8,7 @@ import pathlib
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.utils import tmpfile, tmpdir, dependency_depth
 from dask.dataframe.utils import assert_eq
 
diff --git a/dask/dataframe/io/tests/test_io.py b/dask/dataframe/io/tests/test_io.py
index 6f10b2a6c..6f5fe0462 100644
--- a/dask/dataframe/io/tests/test_io.py
+++ b/dask/dataframe/io/tests/test_io.py
@@ -1,6 +1,5 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 
 import pytest
 from threading import Lock
@@ -8,6 +7,7 @@ from multiprocessing.pool import ThreadPool
 
 import dask.array as da
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.dataframe.io.io import _meta_from_array
 from dask.delayed import Delayed, delayed
 
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 718216bfd..78c38a3fd 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -4,12 +4,12 @@ from distutils.version import LooseVersion
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import dask
 import dask.multiprocessing
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.dataframe.utils import assert_eq, PANDAS_VERSION
 from dask.dataframe.io.parquet.utils import _parse_pandas_metadata
 from dask.dataframe.optimize import optimize_read_parquet_getitem
@@ -2243,3 +2243,17 @@ def test_roundtrip_pandas_chunksize(tmpdir, write_engine, read_engine):
     )
 
     assert_eq(pdf, ddf_read)
+
+
+def test_read_pandas_fastparquet_partitioned(tmpdir, engine):
+    check_fastparquet()
+
+    pdf = pd.DataFrame(
+        [{"str": str(i), "int": i, "group": "ABC"[i % 3]} for i in range(6)]
+    )
+    path = str(tmpdir)
+    pdf.to_parquet(path, partition_cols=["group"], engine="fastparquet")
+    ddf_read = dd.read_parquet(path, engine=engine)
+
+    assert len(ddf_read["group"].compute()) == 6
+    assert len(ddf_read.compute().group) == 6
diff --git a/dask/dataframe/tests/test_accessors.py b/dask/dataframe/tests/test_accessors.py
index b86c7fbb4..aafae2b91 100644
--- a/dask/dataframe/tests/test_accessors.py
+++ b/dask/dataframe/tests/test_accessors.py
@@ -47,7 +47,7 @@ def test_register(obj, registrar):
     with ensure_removed(obj, "mine"):
         before = set(dir(obj))
         registrar("mine")(MyAccessor)
-        instance = dd.from_pandas(obj._partition_type([]), 2)
+        instance = dd.from_pandas(obj._partition_type([], dtype=float), 2)
         assert instance.mine.prop == "item"
         after = set(dir(obj))
         assert (before ^ after) == {"mine"}
diff --git a/dask/dataframe/tests/test_arithmetics_reduction.py b/dask/dataframe/tests/test_arithmetics_reduction.py
index bf4c8affa..6998afc03 100644
--- a/dask/dataframe/tests/test_arithmetics_reduction.py
+++ b/dask/dataframe/tests/test_arithmetics_reduction.py
@@ -7,6 +7,7 @@ import pandas as pd
 from pandas.api.types import is_datetime64_ns_dtype
 
 import dask.dataframe as dd
+from dask.dataframe._compat import PANDAS_GT_100
 from dask.dataframe.utils import (
     assert_eq,
     assert_dask_graph,
@@ -799,17 +800,6 @@ def test_reductions_timedelta(split_every):
     ds = pd.Series(pd.to_timedelta([2, 3, 4, np.nan, 5]))
     dds = dd.from_pandas(ds, 2)
 
-    with pytest.warns(None):
-        # runtime warnings; https://github.com/dask/dask/issues/2381
-        assert_eq(dds.var(split_every=split_every), ds.var())
-        if not PANDAS_GT_0250:
-            # https://github.com/pandas-dev/pandas/issues/18880
-            assert_eq(
-                dds.var(split_every=split_every, skipna=False), ds.var(skipna=False)
-            )
-
-    assert_eq(dds.var(ddof=0, split_every=split_every), ds.var(ddof=0))
-
     assert_eq(dds.sum(split_every=split_every), ds.sum())
     assert_eq(dds.min(split_every=split_every), ds.min())
     assert_eq(dds.max(split_every=split_every), ds.max())
@@ -822,12 +812,12 @@ def test_reductions_timedelta(split_every):
         (
             pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}, index=[0, 1, 3]),
             0,
-            pd.Series([]),
+            pd.Series([], dtype="float64"),
         ),
         (
             pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}, index=[0, 1, 3]),
             1,
-            pd.Series([]),
+            pd.Series([], dtype="float64"),
         ),
         (pd.Series([1, 2.5, 6]), None, None),
     ],
@@ -1028,7 +1018,6 @@ def test_reductions_non_numeric_dtypes():
 
     # ToDo: pandas supports timedelta std, dask returns float64
     # assert_eq(dds.std(), pds.std())
-    assert_eq(dds.var(), pds.var())
 
     # ToDo: pandas supports timedelta std, otherwise dask raises:
     # TypeError: unsupported operand type(s) for *: 'float' and 'Timedelta'
@@ -1147,13 +1136,23 @@ def test_reductions_frame_dtypes():
     df_no_timedelta = df.drop("timedelta", axis=1, inplace=False)
     ddf_no_timedelta = dd.from_pandas(df_no_timedelta, 3)
 
-    assert_eq(df.sum(), ddf.sum())
+    if not PANDAS_GT_100:
+        # https://github.com/pandas-dev/pandas/issues/30886
+        assert_eq(df.sum(), ddf.sum())
+        assert_eq(df_no_timedelta.mean(), ddf_no_timedelta.mean())
+    else:
+        assert_eq(df.drop(columns="dt").sum(), ddf.drop(columns="dt").sum())
+        assert_eq(
+            df_no_timedelta.drop(columns="dt").mean(),
+            ddf_no_timedelta.drop(columns="dt").mean(),
+        )
+
     assert_eq(df.prod(), ddf.prod())
     assert_eq(df.min(), ddf.min())
     assert_eq(df.max(), ddf.max())
     assert_eq(df.count(), ddf.count())
     assert_eq(df_no_timedelta.std(), ddf_no_timedelta.std())
-    assert_eq(df.var(), ddf.var())
+    assert_eq(df_no_timedelta.var(), ddf_no_timedelta.var())
     if PANDAS_GT_0250:
         # https://github.com/pandas-dev/pandas/issues/18880
         assert_eq(
@@ -1176,8 +1175,6 @@ def test_reductions_frame_dtypes():
         assert_eq(df.var(ddof=0, skipna=False), ddf.var(ddof=0, skipna=False))
     assert_eq(df.sem(ddof=0), ddf.sem(ddof=0))
 
-    assert_eq(df_no_timedelta.mean(), ddf_no_timedelta.mean())
-
     assert_eq(df._get_numeric_data(), ddf._get_numeric_data())
 
     numerics = ddf[["int", "float"]]
diff --git a/dask/dataframe/tests/test_categorical.py b/dask/dataframe/tests/test_categorical.py
index be9ca91d1..07553c9e1 100644
--- a/dask/dataframe/tests/test_categorical.py
+++ b/dask/dataframe/tests/test_categorical.py
@@ -2,11 +2,12 @@ import operator
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
+from dask.dataframe import _compat
 from dask.dataframe.core import _concat
 from dask.dataframe.utils import (
     make_meta,
@@ -213,7 +214,7 @@ def test_categorize():
 
 def test_categorize_index():
     # Object dtype
-    ddf = dd.from_pandas(tm.makeDataFrame(), npartitions=5)
+    ddf = dd.from_pandas(_compat.makeDataFrame(), npartitions=5)
     df = ddf.compute()
 
     ddf2 = ddf.categorize()
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index bae65776b..24f4408a2 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -5,13 +5,14 @@ from operator import add
 import pytest
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 from pandas.io.formats import format as pandas_format
 
 import dask
 import dask.array as da
 from dask.array.numpy_compat import _numpy_118
 import dask.dataframe as dd
+from dask.dataframe import _compat
+from dask.dataframe._compat import tm, PANDAS_GT_0230, PANDAS_GT_100
 from dask.base import compute_as_if_collection
 from dask.utils import put_lines, M
 
@@ -177,7 +178,7 @@ def test_attributes():
     assert "a" in dir(df)
     assert 5 not in dir(df)
 
-    df = dd.from_pandas(tm.makeTimeDataFrame(), npartitions=3)
+    df = dd.from_pandas(_compat.makeTimeDataFrame(), npartitions=3)
     pytest.raises(AttributeError, lambda: df.foo)
 
 
@@ -1542,7 +1543,7 @@ def test_dataframe_picklable():
     cloudpickle = pytest.importorskip("cloudpickle")
     cp_dumps = cloudpickle.dumps
 
-    d = tm.makeTimeDataFrame()
+    d = _compat.makeTimeDataFrame()
     df = dd.from_pandas(d, npartitions=3)
     df = df + 2
 
@@ -1926,7 +1927,7 @@ def test_embarrassingly_parallel_operations():
 
 
 def test_fillna():
-    df = tm.makeMissingDataframe(0.8, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
 
     assert_eq(ddf.fillna(100), df.fillna(100))
@@ -1958,7 +1959,8 @@ def test_fillna():
     pytest.raises(NotImplementedError, lambda: ddf.fillna(0, limit=10))
     pytest.raises(NotImplementedError, lambda: ddf.fillna(0, limit=10, axis=1))
 
-    df = tm.makeMissingDataframe(0.2, 42)
+    df = _compat.makeMissingDataframe()
+    df.iloc[:15, 0] = np.nan  # all NaN partition
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
     pytest.raises(ValueError, lambda: ddf.fillna(method="pad").compute())
     assert_eq(df.fillna(method="pad", limit=3), ddf.fillna(method="pad", limit=3))
@@ -1975,7 +1977,7 @@ def test_fillna_duplicate_index():
 
 
 def test_fillna_multi_dataframe():
-    df = tm.makeMissingDataframe(0.8, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
 
     assert_eq(ddf.A.fillna(ddf.B), df.A.fillna(df.B))
@@ -1983,7 +1985,7 @@ def test_fillna_multi_dataframe():
 
 
 def test_ffill_bfill():
-    df = tm.makeMissingDataframe(0.8, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
 
     assert_eq(ddf.ffill(), df.ffill())
@@ -2115,14 +2117,18 @@ def test_select_dtypes(include, exclude):
 
     tm.assert_series_equal(result.dtypes.value_counts(), expected.dtypes.value_counts())
 
-    if PANDAS_VERSION >= "0.23.0":
-        ctx = pytest.warns(FutureWarning)
-    else:
-        ctx = pytest.warns(None)
-
-    with ctx:
-        tm.assert_series_equal(a.get_ftype_counts(), df.get_ftype_counts())
-        tm.assert_series_equal(result.get_ftype_counts(), expected.get_ftype_counts())
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        if PANDAS_GT_0230:
+            ctx = pytest.warns(FutureWarning)
+        else:
+            ctx = pytest.warns(None)
+
+        with ctx:
+            tm.assert_series_equal(a.get_ftype_counts(), df.get_ftype_counts())
+            tm.assert_series_equal(
+                result.get_ftype_counts(), expected.get_ftype_counts()
+            )
 
 
 def test_deterministic_apply_concat_apply_names():
@@ -2604,7 +2610,7 @@ def test_round():
 
 def test_cov():
     # DataFrame
-    df = pd.util.testing.makeMissingDataframe(0.3, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=6)
 
     res = ddf.cov()
@@ -2646,7 +2652,7 @@ def test_cov():
 
 def test_corr():
     # DataFrame
-    df = pd.util.testing.makeMissingDataframe(0.3, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=6)
 
     res = ddf.corr()
@@ -2694,7 +2700,7 @@ def test_corr():
 def test_corr_same_name():
     # Series with same names (see https://github.com/dask/dask/issues/4906)
 
-    df = pd.util.testing.makeMissingDataframe(0.3, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=6)
 
     result = ddf.A.corr(ddf.B.rename("A"))
@@ -2807,7 +2813,7 @@ def test_apply_infer_columns():
 
 
 def test_index_time_properties():
-    i = tm.makeTimeSeries()
+    i = _compat.makeTimeSeries()
     a = dd.from_pandas(i, npartitions=3)
 
     assert "day" in dir(a.index)
@@ -3027,6 +3033,7 @@ def _assert_info(df, ddf, memory_usage=True):
     stdout_pd = buf_pd.getvalue()
     stdout_da = buf_da.getvalue()
     stdout_da = stdout_da.replace(str(type(ddf)), str(type(df)))
+    # TODO
     assert stdout_pd == stdout_da
 
 
@@ -3137,7 +3144,7 @@ def test_gh_1301():
 
 
 def test_timeseries_sorted():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df.reset_index(), npartitions=2)
     df.index.name = "index"
     assert_eq(ddf.set_index("index", sorted=True, drop=True), df)
@@ -3369,7 +3376,7 @@ def test_diff():
 
 
 def test_shift():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df, npartitions=4)
 
     # DataFrame
@@ -3392,8 +3399,8 @@ def test_shift():
 
 @pytest.mark.parametrize("data_freq,divs1", [("B", False), ("D", True), ("H", True)])
 def test_shift_with_freq_DatetimeIndex(data_freq, divs1):
-    df = tm.makeTimeDataFrame(30)
-    df = df.set_index(tm.makeDateIndex(30, freq=data_freq))
+    df = _compat.makeTimeDataFrame()
+    df = df.set_index(_compat.makeDateIndex(30, freq=data_freq))
     ddf = dd.from_pandas(df, npartitions=4)
     for freq, divs2 in [("S", True), ("W", False), (pd.Timedelta(10, unit="h"), True)]:
         for d, p in [(ddf, df), (ddf.A, df.A), (ddf.index, df.index)]:
@@ -3408,7 +3415,7 @@ def test_shift_with_freq_DatetimeIndex(data_freq, divs1):
 
 @pytest.mark.parametrize("data_freq,divs", [("B", False), ("D", True), ("H", True)])
 def test_shift_with_freq_PeriodIndex(data_freq, divs):
-    df = tm.makeTimeDataFrame(30)
+    df = _compat.makeTimeDataFrame()
     # PeriodIndex
     df = df.set_index(pd.period_range("2000-01-01", periods=30, freq=data_freq))
     ddf = dd.from_pandas(df, npartitions=4)
@@ -3421,16 +3428,16 @@ def test_shift_with_freq_PeriodIndex(data_freq, divs):
     assert_eq(res, df.index.shift(2))
     assert res.known_divisions == divs
 
-    df = tm.makeTimeDataFrame(30)
+    df = _compat.makeTimeDataFrame()
     with pytest.raises(ValueError):
         ddf.index.shift(2, freq="D")  # freq keyword not supported
 
 
 def test_shift_with_freq_TimedeltaIndex():
-    df = tm.makeTimeDataFrame(30)
+    df = _compat.makeTimeDataFrame()
     # TimedeltaIndex
     for data_freq in ["T", "D", "H"]:
-        df = df.set_index(tm.makeTimedeltaIndex(30, freq=data_freq))
+        df = df.set_index(_compat.makeTimedeltaIndex(30, freq=data_freq))
         ddf = dd.from_pandas(df, npartitions=4)
         for freq in ["S", pd.Timedelta(10, unit="h")]:
             for d, p in [(ddf, df), (ddf.A, df.A), (ddf.index, df.index)]:
@@ -3445,7 +3452,7 @@ def test_shift_with_freq_TimedeltaIndex():
 
 def test_shift_with_freq_errors():
     # Other index types error
-    df = tm.makeDataFrame()
+    df = _compat.makeDataFrame()
     ddf = dd.from_pandas(df, npartitions=4)
     pytest.raises(NotImplementedError, lambda: ddf.shift(2, freq="S"))
     pytest.raises(NotImplementedError, lambda: ddf.A.shift(2, freq="S"))
@@ -3468,9 +3475,20 @@ def test_first_and_last(method):
             assert_eq(f(ddf.A, offset), f(df.A, offset))
 
 
+xfail_hash_object = pytest.mark.xfail(PANDAS_GT_100, reason="GH-30887")
+
+
 @pytest.mark.parametrize("npartitions", [1, 4, 20])
 @pytest.mark.parametrize("split_every", [2, 5])
-@pytest.mark.parametrize("split_out", [None, 1, 5, 20])
+@pytest.mark.parametrize(
+    "split_out",
+    [
+        None,
+        1,
+        pytest.param(5, marks=xfail_hash_object),
+        pytest.param(20, marks=xfail_hash_object),
+    ],
+)
 def test_hash_split_unique(npartitions, split_every, split_out):
     from string import ascii_lowercase
 
@@ -3987,13 +4005,13 @@ def test_scalar_with_array():
 
 def test_has_parallel_type():
     assert has_parallel_type(pd.DataFrame())
-    assert has_parallel_type(pd.Series())
+    assert has_parallel_type(pd.Series(dtype=float))
     assert not has_parallel_type(123)
 
 
 def test_meta_error_message():
     with pytest.raises(TypeError) as info:
-        dd.DataFrame({("x", 1): 123}, "x", pd.Series(), [None, None])
+        dd.DataFrame({("x", 1): 123}, "x", pd.Series(dtype=float), [None, None])
 
     assert "Series" in str(info.value)
     assert "DataFrame" in str(info.value)
@@ -4075,6 +4093,9 @@ def test_dtype_cast():
 @pytest.mark.parametrize("map_npart", [1, 3])
 @pytest.mark.parametrize("sorted_index", [False, True])
 @pytest.mark.parametrize("sorted_map_index", [False, True])
+@pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
 def test_series_map(base_npart, map_npart, sorted_index, sorted_map_index):
     base = pd.Series(
         ["".join(np.random.choice(["a", "b", "c"], size=3)) for x in range(100)]
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 7e28464c2..3ef7cf017 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -4,12 +4,13 @@ from packaging import version
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 
 import pytest
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm, PANDAS_GT_100
+from dask.dataframe import _compat
 from dask.dataframe.utils import (
     assert_eq,
     assert_dask_graph,
@@ -475,7 +476,7 @@ def test_series_groupby_errors():
 
 
 def test_groupby_index_array():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df, npartitions=2)
 
     # first select column, then group
@@ -494,7 +495,7 @@ def test_groupby_index_array():
 
 
 def test_groupby_set_index():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df, npartitions=2)
     pytest.raises(TypeError, lambda: ddf.groupby(df.index.month, as_index=False))
 
@@ -681,7 +682,10 @@ def test_split_apply_combine_on_series(empty):
     pytest.raises(KeyError, lambda: ddf.groupby("x"))
     pytest.raises(KeyError, lambda: ddf.groupby(["a", "x"]))
     pytest.raises(KeyError, lambda: ddf.groupby("a")["x"])
-    pytest.raises(KeyError, lambda: ddf.groupby("a")["b", "x"])
+    with warnings.catch_warnings():
+        # pandas warns about using tuples before throwing the KeyError
+        warnings.simplefilter("ignore", FutureWarning)
+        pytest.raises(KeyError, lambda: ddf.groupby("a")["b", "x"])
     pytest.raises(KeyError, lambda: ddf.groupby("a")[["b", "x"]])
 
     # test graph node labels
@@ -878,7 +882,7 @@ def test_numeric_column_names():
 
 
 def test_groupby_apply_tasks():
-    df = pd.util.testing.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     df["A"] = df.A // 0.1
     df["B"] = df.B // 0.1
     ddf = dd.from_pandas(df, npartitions=10)
@@ -974,11 +978,13 @@ def test_aggregate__examples(spec, split_every, grouper):
     # Warning from pandas deprecation .agg(dict[dict])
     # it's from pandas, so no reason to assert the deprecation warning,
     # but we should still test it for now
-    with pytest.warns(None):
-        assert_eq(
-            pdf.groupby(grouper(pdf)).agg(spec),
-            ddf.groupby(grouper(ddf)).agg(spec, split_every=split_every),
-        )
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(None):
+            assert_eq(
+                pdf.groupby(grouper(pdf)).agg(spec),
+                ddf.groupby(grouper(ddf)).agg(spec, split_every=split_every),
+            )
 
 
 @pytest.mark.parametrize(
@@ -1013,11 +1019,13 @@ def test_series_aggregate__examples(spec, split_every, grouper):
     # Warning from pandas deprecation .agg(dict[dict])
     # it's from pandas, so no reason to assert the deprecation warning,
     # but we should still test it for now
-    with pytest.warns(None):
-        assert_eq(
-            ps.groupby(grouper(pdf)).agg(spec),
-            ds.groupby(grouper(ddf)).agg(spec, split_every=split_every),
-        )
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(None):
+            assert_eq(
+                ps.groupby(grouper(pdf)).agg(spec),
+                ds.groupby(grouper(ddf)).agg(spec, split_every=split_every),
+            )
 
 
 def test_aggregate__single_element_groups(agg_func):
diff --git a/dask/dataframe/tests/test_hashing.py b/dask/dataframe/tests/test_hashing.py
index ca37bbb1a..d046efb62 100644
--- a/dask/dataframe/tests/test_hashing.py
+++ b/dask/dataframe/tests/test_hashing.py
@@ -1,7 +1,8 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import dask.dataframe as dd
+from dask.dataframe import _compat
+from dask.dataframe._compat import tm, PANDAS_GT_100
 from pandas.util import hash_pandas_object
 
 import pytest
@@ -20,11 +21,11 @@ from dask.dataframe.utils import assert_eq
         pd.Index([1, 2, 3]),
         pd.Index([True, False, True]),
         pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}),
-        pd.util.testing.makeMissingDataframe(),
-        pd.util.testing.makeMixedDataFrame(),
-        pd.util.testing.makeTimeDataFrame(),
-        pd.util.testing.makeTimeSeries(),
-        pd.util.testing.makeTimedeltaIndex(),
+        _compat.makeMissingDataframe(),
+        _compat.makeMixedDataFrame(),
+        _compat.makeTimeDataFrame(),
+        _compat.makeTimeSeries(),
+        _compat.makeTimedeltaIndex(),
     ],
 )
 def test_hash_pandas_object(obj):
@@ -64,16 +65,24 @@ def test_object_missing_values():
     tm.assert_series_equal(h1, h2)
 
 
+xfail_hash_object = pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
+
+
 @pytest.mark.parametrize(
     "obj",
     [
         pd.Index([1, 2, 3]),
-        pd.Index([True, False, True], index=[1.5, 1.1, 3.3]),
+        pytest.param(pd.Index([True, False, True]), marks=xfail_hash_object),
         pd.Series([1, 2, 3]),
         pd.Series([1.0, 1.5, 3.2]),
         pd.Series([1.0, 1.5, 3.2], index=[1.5, 1.1, 3.3]),
         pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}),
-        pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}, index=["a", "z", "x"]),
+        pytest.param(
+            pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}, index=["a", "z", "x"]),
+            marks=xfail_hash_object,
+        ),
     ],
 )
 def test_hash_object_dispatch(obj):
diff --git a/dask/dataframe/tests/test_indexing.py b/dask/dataframe/tests/test_indexing.py
index 68cf5b7e7..3c10de580 100644
--- a/dask/dataframe/tests/test_indexing.py
+++ b/dask/dataframe/tests/test_indexing.py
@@ -1,11 +1,11 @@
 import pandas as pd
-import pandas.util.testing as tm
 import numpy as np
 
 import pytest
 
 import dask.dataframe as dd
 
+from dask.dataframe._compat import tm, PANDAS_GT_100
 from dask.dataframe.indexing import _coerce_loc_index
 from dask.dataframe.utils import assert_eq, make_meta, PANDAS_VERSION
 
@@ -37,24 +37,30 @@ def test_loc():
     else:
         expected_warning = None
 
-    with pytest.warns(expected_warning):
-        assert_eq(d.loc[[3, 4, 1, 8]], full.loc[[3, 4, 1, 8]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.loc[[3, 4, 1, 9]], full.loc[[3, 4, 1, 9]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.loc[np.array([3, 4, 1, 9])], full.loc[np.array([3, 4, 1, 9])])
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(expected_warning):
+            assert_eq(d.loc[[3, 4, 1, 8]], full.loc[[3, 4, 1, 8]])
+        with pytest.warns(expected_warning):
+            assert_eq(d.loc[[3, 4, 1, 9]], full.loc[[3, 4, 1, 9]])
+        with pytest.warns(expected_warning):
+            assert_eq(d.loc[np.array([3, 4, 1, 9])], full.loc[np.array([3, 4, 1, 9])])
 
     assert_eq(d.a.loc[5], full.a.loc[5:5])
     assert_eq(d.a.loc[3:8], full.a.loc[3:8])
     assert_eq(d.a.loc[:8], full.a.loc[:8])
     assert_eq(d.a.loc[3:], full.a.loc[3:])
     assert_eq(d.a.loc[[5]], full.a.loc[[5]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.a.loc[[3, 4, 1, 8]], full.a.loc[[3, 4, 1, 8]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.a.loc[[3, 4, 1, 9]], full.a.loc[[3, 4, 1, 9]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.a.loc[np.array([3, 4, 1, 9])], full.a.loc[np.array([3, 4, 1, 9])])
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(expected_warning):
+            assert_eq(d.a.loc[[3, 4, 1, 8]], full.a.loc[[3, 4, 1, 8]])
+        with pytest.warns(expected_warning):
+            assert_eq(d.a.loc[[3, 4, 1, 9]], full.a.loc[[3, 4, 1, 9]])
+        with pytest.warns(expected_warning):
+            assert_eq(
+                d.a.loc[np.array([3, 4, 1, 9])], full.a.loc[np.array([3, 4, 1, 9])]
+            )
     assert_eq(d.a.loc[[]], full.a.loc[[]])
     assert_eq(d.a.loc[np.array([])], full.a.loc[np.array([])])
 
@@ -82,8 +88,8 @@ def test_loc_non_informative_index():
 
 
 def test_loc_with_text_dates():
-    A = tm.makeTimeSeries(10).iloc[:5]
-    B = tm.makeTimeSeries(10).iloc[5:]
+    A = dd._compat.makeTimeSeries().iloc[:5]
+    B = dd._compat.makeTimeSeries().iloc[5:]
     s = dd.Series(
         {("df", 0): A, ("df", 1): B},
         "df",
@@ -168,6 +174,7 @@ def test_loc2d():
         d.a.loc[d.a % 2 == 0, 3]
 
 
+@pytest.mark.skip(PANDAS_GT_100, reason="Removed in pandas 1.0")
 def test_loc2d_some_missing():
     with pytest.warns(FutureWarning):
         assert_eq(d.loc[[3, 4, 3], ["a"]], full.loc[[3, 4, 3], ["a"]])
diff --git a/dask/dataframe/tests/test_multi.py b/dask/dataframe/tests/test_multi.py
index 34e67f6a6..e79e98c60 100644
--- a/dask/dataframe/tests/test_multi.py
+++ b/dask/dataframe/tests/test_multi.py
@@ -4,9 +4,9 @@ import dask
 import dask.dataframe as dd
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 
 from dask.base import compute_as_if_collection
+from dask.dataframe._compat import tm, PANDAS_GT_100
 from dask.dataframe.core import _Frame
 from dask.dataframe.methods import concat, concat_kwargs
 from dask.dataframe.multi import (
@@ -221,7 +221,8 @@ def list_eq(aa, bb):
     else:
         av = a.sort_values().values
         bv = b.sort_values().values
-    tm.assert_numpy_array_equal(av, bv)
+
+    dd._compat.assert_numpy_array_equal(av, bv)
 
 
 @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
@@ -1971,6 +1972,9 @@ def test_multi_duplicate_divisions():
     assert_eq(r1, r2)
 
 
+@pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
 def test_merge_outer_empty():
     # Issue #5470 bug reproducer
     k_clusters = 3
diff --git a/dask/dataframe/tests/test_reshape.py b/dask/dataframe/tests/test_reshape.py
index 910cdccfd..829663ef4 100644
--- a/dask/dataframe/tests/test_reshape.py
+++ b/dask/dataframe/tests/test_reshape.py
@@ -1,10 +1,10 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import dask.dataframe as dd
 
+from dask.dataframe._compat import tm
 from dask.dataframe.utils import assert_eq, make_meta, PANDAS_VERSION, PANDAS_GT_0240
 
 
diff --git a/dask/dataframe/tests/test_shuffle.py b/dask/dataframe/tests/test_shuffle.py
index f80ae0b35..91ecb4087 100644
--- a/dask/dataframe/tests/test_shuffle.py
+++ b/dask/dataframe/tests/test_shuffle.py
@@ -9,10 +9,10 @@ import numpy as np
 import string
 import multiprocessing as mp
 from copy import copy
-import pandas.util.testing as tm
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm, PANDAS_GT_100, assert_categorical_equal
 from dask import delayed
 from dask.base import compute_as_if_collection
 from dask.dataframe.shuffle import (
@@ -142,6 +142,9 @@ def test_partitioning_index():
     assert len(np.unique(res)) > 1
 
 
+@pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
 def test_partitioning_index_categorical_on_values():
     df = pd.DataFrame({"a": list(string.ascii_letters), "b": [1, 2, 3, 4] * 13})
     df.a = df.a.astype("category")
@@ -752,7 +755,7 @@ def test_set_index_categorical():
 
     # sorted with the metric defined by the Categorical
     divisions = pd.Categorical(result.divisions, dtype=dtype)
-    tm.assert_categorical_equal(divisions, divisions.sort_values())
+    assert_categorical_equal(divisions, divisions.sort_values())
 
 
 def test_compute_divisions():
diff --git a/dask/dataframe/tests/test_ufunc.py b/dask/dataframe/tests/test_ufunc.py
index 74c6de606..a74bddaa4 100644
--- a/dask/dataframe/tests/test_ufunc.py
+++ b/dask/dataframe/tests/test_ufunc.py
@@ -1,7 +1,6 @@
 import pytest
 
 pd = pytest.importorskip("pandas")
-import pandas.util.testing as tm
 
 import numpy as np
 
@@ -183,10 +182,10 @@ def test_ufunc_array_wrap(ufunc):
     assert_eq(dafunc(ds), pd.Series(npfunc(s), index=s.index))
 
     assert isinstance(npfunc(ds), np.ndarray)
-    tm.assert_numpy_array_equal(npfunc(ds), npfunc(s))
+    np.testing.assert_equal(npfunc(ds), npfunc(s))
 
     assert isinstance(dafunc(s), np.ndarray)
-    tm.assert_numpy_array_equal(dafunc(s), npfunc(s))
+    np.testing.assert_array_equal(dafunc(s), npfunc(s))
 
     df = pd.DataFrame(
         {
@@ -205,10 +204,10 @@ def test_ufunc_array_wrap(ufunc):
     assert_eq(dafunc(ddf), exp)
 
     assert isinstance(npfunc(ddf), np.ndarray)
-    tm.assert_numpy_array_equal(npfunc(ddf), npfunc(df))
+    np.testing.assert_array_equal(npfunc(ddf), npfunc(df))
 
     assert isinstance(dafunc(df), np.ndarray)
-    tm.assert_numpy_array_equal(dafunc(df), npfunc(df))
+    np.testing.assert_array_equal(dafunc(df), npfunc(df))
 
 
 _UFUNCS_2ARG = [
@@ -453,7 +452,7 @@ def test_2args_with_array(ufunc, pandas, darray):
     assert isinstance(dafunc(dask, darray), dask_type)
     assert isinstance(dafunc(darray, dask), dask_type)
 
-    tm.assert_numpy_array_equal(
+    np.testing.assert_array_equal(
         dafunc(dask, darray).compute().values, npfunc(pandas.values, darray).compute()
     )
 
@@ -461,10 +460,10 @@ def test_2args_with_array(ufunc, pandas, darray):
     assert isinstance(npfunc(dask, darray), dask_type)
     assert isinstance(npfunc(darray, dask), dask_type)
 
-    tm.assert_numpy_array_equal(
+    np.testing.assert_array_equal(
         npfunc(dask, darray).compute().values, npfunc(pandas.values, darray.compute())
     )
-    tm.assert_numpy_array_equal(
+    np.testing.assert_array_equal(
         npfunc(darray, dask).compute().values, npfunc(darray.compute(), pandas.values)
     )
 
diff --git a/dask/dataframe/tests/test_utils_dataframe.py b/dask/dataframe/tests/test_utils_dataframe.py
index 764a18251..103cc1b9f 100644
--- a/dask/dataframe/tests/test_utils_dataframe.py
+++ b/dask/dataframe/tests/test_utils_dataframe.py
@@ -2,8 +2,8 @@ import re
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.dataframe.core import apply_and_enforce
 from dask.dataframe.utils import (
     shard_df_on_index,