code: def test_dataframe_groupby_tasks(client):
      df = make_time_dataframe()
  
      df["A"] = df.A // 0.1
      df["B"] = df.B // 0.1
      ddf = dd.from_pandas(df, npartitions=10)
  
      for ind in [lambda x: "A", lambda x: x.A]:
          a = df.groupby(ind(df)).apply(len)
          b = ddf.groupby(ind(ddf)).apply(len, meta=(None, int))
          assert_equal(a, b.compute().sort_index())
          assert not any("partd" in k[0] for k in b.dask)
  
          a = df.groupby(ind(df)).B.apply(len)
          b = ddf.groupby(ind(ddf)).B.apply(len, meta=("B", int))
          assert_equal(a, b.compute().sort_index())
          assert not any("partd" in k[0] for k in b.dask)
  
      with pytest.raises((NotImplementedError, ValueError)):
          ddf.groupby(ddf[["A", "B"]]).apply(len, meta=int)
  
      a = df.groupby(["A", "B"]).apply(len)
      b = ddf.groupby(["A", "B"]).apply(len, meta=(None, int))
  
      assert_equal(a, b.compute().sort_index())
  
file_path: distributed/distributed/tests/test_dask_collections.py
imports: from __future__ import annotations
  import pytest
  from packaging.version import parse
  from pandas.testing import assert_frame_equal, assert_index_equal, assert_series_equal
  import dask
  import dask.bag
  import dask.dataframe
  from distributed.client import wait
  from distributed.nanny import Nanny
  from distributed.utils_test import gen_cluster
label_desc: Use image node_id 6 for calling a global function with example usage: test_dataframe_groupby_tasks(client) without return types