code: def persist_dask_df_equal_parts_per_worker(
      dask_df, client, return_type="dask_cudf.DataFrame"
  ):
      """
      Persist dask_df with equal parts per worker
      Args:
          dask_df: dask_cudf.DataFrame
          client: dask.distributed.Client
          return_type: str, "dask_cudf.DataFrame" or "dict"
      Returns:
          persisted_keys: dict of {worker: [persisted_keys]}
      """
      if return_type not in ["dask_cudf.DataFrame", "dict"]:
          raise ValueError(
              "return_type must be either 'dask_cudf.DataFrame' or 'dict'"
          )
  
      ddf_keys = dask_df.to_delayed()
      workers = client.scheduler_info()["workers"].keys()
      ddf_keys_ls = _chunk_lst(ddf_keys, len(workers))
      persisted_keys_d = {}
      for w, ddf_k in zip(workers, ddf_keys_ls):
          persisted_keys_d[w] = client.compute(
              ddf_k, workers=w, allow_other_workers=False, pure=False
          )
  
      persisted_keys_ls = [
          item
          for sublist in persisted_keys_d.values()
          for item in sublist
      ]
      wait(persisted_keys_ls)
      if return_type == "dask_cudf.DataFrame":
          dask_df = dask_cudf.from_delayed(
              persisted_keys_ls, meta=dask_df._meta
          ).persist()
          wait(dask_df)
          return dask_df
  
      return persisted_keys_d
  
imports: import numpy
  from dask.distributed import futures_of, default_client, wait
  from toolz import first
  import collections
  import dask_cudf
  from dask.array.core import Array
  from dask_cudf.core import DataFrame
  from dask_cudf.core import Series
  from functools import reduce
  import cugraph.dask.comms.comms
  from dask.delayed import delayed
  import cudf
label_desc: Use image node_id 8 for calling a global function with example usage: persist_dask_df_equal_parts_per_worker(dask_df, client, return_type) and returns: persisted_keys_d, dask_df