code: def _create_df_from_edge_ar(src_ar, dst_ar, single_gpu=True):
      if not single_gpu:
          nworkers = len(get_client().scheduler_info()["workers"])
          npartitions = nworkers * 1
      if single_gpu:
          df = cudf.DataFrame(data={src_n: src_ar, dst_n: dst_ar})
      else:
          if isinstance(src_ar, cp.ndarray):
              src_ar = src_ar.get()
          if isinstance(dst_ar, cp.ndarray):
              dst_ar = dst_ar.get()
  
          df = pd.DataFrame(data={src_n: src_ar, dst_n: dst_ar})
          # Only save stuff in host memory
          df = dd.from_pandas(df, npartitions=npartitions).persist()
          df = df.map_partitions(cudf.DataFrame.from_pandas)
  
      df = df.reset_index(drop=True)
      return df
  
file_path: cugraph/python/cugraph-dgl/cugraph_dgl/utils/cugraph_conversion_utils.py
imports: from __future__ import annotations
  from typing import Dict, Tuple, Union
  import cudf
  import pandas
  import dask.dataframe
  import dask_cudf
  from dask.distributed import get_client
  import cupy
  from cugraph.utilities.utils import import_optional
  from cugraph.gnn.dgl_extensions.dgl_uniform_sampler import src_n, dst_n
label_desc: Use image node_id 4 for calling a global function with example usage: _create_df_from_edge_ar(src_ar, dst_ar, single_gpu) and returns: df