class_docstr: Class to centralize distributed data management. Functionalities include:
  - Data colocation
  - Worker information extraction
  - GPU futures extraction,
  
  Additional functionality can be added as needed. This class **does not**
  contain the actual data, just the metadata necessary to handle it,
  including common pieces of code that need to be performed to call
  Dask functions.
  
  The constructor is not meant to be used directly, but through the factory
  method DistributedDataHandler.create
code: def calculate_parts_to_sizes(self, comms=None, ranks=None):
      if self.worker_info is None and comms is not None:
          self.calculate_worker_and_rank_info(comms)
  
      self.total_rows = 0
  
      self.parts_to_sizes = dict()
  
      parts = [
          (
              wf[0],
              self.client.submit(
                  _get_rows,
                  wf[1],
                  self.multiple,
                  workers=[wf[0]],
                  pure=False,
              ),
          )
          for idx, wf in enumerate(self.worker_to_parts.items())
      ]
  
      sizes = self.client.compute(parts, sync=True)
  
      for w, sizes_parts in sizes:
          sizes, total = sizes_parts
          self.parts_to_sizes[self.worker_info[w]["rank"]] = sizes
  
          self.total_rows += total
  
imports: from collections.abc import Sequence
  from collections import OrderedDict
  from dask_cudf.core import DataFrame
  from dask_cudf.core import Series
  import cugraph.dask.comms.comms
  from cugraph.dask.common.read_utils import MissingUCXPy
  from cugraph.dask.common.part_utils import _extract_partitions
  from dask.distributed import default_client
  from toolz import first
  from functools import reduce
label_desc: Use image node_id 5 for calling the DistributedDataHandler obj's underlying member method code with example usage: obj.calculate_parts_to_sizes(comms, ranks) without return types