code: def persist_distributed_data(dask_df, client):
      client = default_client() if client is None else client
      worker_addresses = Comms.get_workers()
      _keys = dask_df.__dask_keys__()
      worker_dict = {}
      for i, key in enumerate(_keys):
          worker_dict[key] = tuple([worker_addresses[i]])
      persisted = client.persist(dask_df, workers=worker_dict)
      parts = futures_of(persisted)
      return parts
  
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 4 for calling a global function with example usage: persist_distributed_data(dask_df, client) and returns: parts