code: def get_length_of_parts(persisted_keys_d, client):
      """
      Get the length of each partition
      Args:
          persisted_keys_d: dict of {worker: [persisted_keys]}
          client: dask.distributed.Client
      Returns:
          length_of_parts: dict of {worker: [length_of_parts]}
      """
      length_of_parts = {}
      for w, p_keys in persisted_keys_d.items():
          length_of_parts[w] = [
              client.submit(
                  len,
                  p_key,
                  pure=False,
                  workers=[w],
                  allow_other_workers=False,
              )
              for p_key in p_keys
          ]
  
      for w, len_futures in length_of_parts.items():
          length_of_parts[w] = client.gather(len_futures)
      return length_of_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 9 for calling a global function with example usage: get_length_of_parts(persisted_keys_d, client) and returns: length_of_parts