code: def get_persisted_df_worker_map(dask_df, client):
      ddf_keys = futures_of(dask_df)
      output_map = {}
      for w, w_keys in client.has_what().items():
          output_map[w] = [
              ddf_k for ddf_k in ddf_keys if ddf_k.key in w_keys
          ]
          if len(output_map[w]) == 0:
              output_map[w] = _create_empty_dask_df_future(
                  dask_df._meta, client, w
              )
      return output_map
  
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 6 for calling a global function with example usage: get_persisted_df_worker_map(dask_df, client) and returns: output_map