code: def _mg_call_plc_betweenness_centrality(
      input_graph,
      client: dask.distributed.client.Client,
      sID: bytes,
      k: dict,
      random_state: int,
      normalized: bool,
      do_expensive_check: bool,
      endpoints: bool = False,
      edge_bc: bool = False,
  ) -> dask_cudf.DataFrame:
      result = [
          client.submit(
              _call_plc_betweenness_centrality,
              input_graph._plc_graph[w],
              sID,
              k if isinstance(k, (int, type(None))) else k[w][0],
              hash((random_state, i)),
              normalized,
              endpoints,
              do_expensive_check,
              edge_bc,
              workers=[w],
              allow_other_workers=False,
              pure=False,
          )
          for i, w in enumerate(Comms.get_workers())
      ]
  
      wait(result)
  
      ddf = dask_cudf.from_delayed(result, verify_meta=False).persist()
      wait(ddf)
      wait([r.release() for r in result])
      return ddf
  
file_path: cugraph/python/cugraph/cugraph/dask/centrality/betweenness_centrality.py
imports: from dask.distributed import wait, get_client
  from pylibcugraph import ResourceHandle, betweenness_centrality, edge_betweenness_centrality
  import cugraph.dask.comms.comms
  from cugraph.dask.common.input_utils import get_distributed_data
  import dask_cudf
  import cudf
  import cupy
  import warnings
  import dask
  from typing import Union
label_desc: Use image node_id 3 for calling a global function with example usage: _mg_call_plc_betweenness_centrality(input_graph, client, sID, k, random_state, normalized, do_expensive_check, endpoints, edge_bc) and returns: ddf