code: def __from_edgelist(
      self,
      input_ddf,
      source="source",
      destination="destination",
      edge_attr=None,
      weight=None,
      edge_id=None,
      edge_type=None,
      renumber=True,
      store_transposed=False,
      legacy_renum_only=False,
  ):
      if not isinstance(input_ddf, dask_cudf.DataFrame):
          raise TypeError("input should be a dask_cudf dataFrame")
  
      if renumber is False:
          if type(source) is list and type(destination) is list:
              raise ValueError(
                  "set renumber to True for multi column ids"
              )
          elif input_ddf[source].dtype not in [
              np.int32,
              np.int64,
          ] or input_ddf[destination].dtype not in [np.int32, np.int64]:
              raise ValueError(
                  "set renumber to True for non integer columns ids"
              )
  
      s_col = source
      d_col = destination
      if not isinstance(s_col, list):
          s_col = [s_col]
      if not isinstance(d_col, list):
          d_col = [d_col]
      if not (
          set(s_col).issubset(set(input_ddf.columns))
          and set(d_col).issubset(set(input_ddf.columns))
      ):
          raise ValueError(
              "source column names and/or destination column "
              "names not found in input. Recheck the source "
              "and destination parameters"
          )
      ddf_columns = s_col + d_col
      self.vertex_columns = ddf_columns.copy()
      _client = default_client()
      workers = _client.scheduler_info()["workers"]
      # Repartition to 2 partitions per GPU for memory efficient process
      input_ddf = input_ddf.repartition(npartitions=len(workers) * 2)
      # The dataframe will be symmetrized iff the graph is undirected
      # otherwise, the inital dataframe will be returned
      if edge_attr is not None:
          if (
              weight is not None
              or edge_id is not None
              or edge_type is not None
          ):
              raise ValueError(
                  "If specifying edge_attr, cannot specify weight/edge_id/edge_type"
              )
          if isinstance(edge_attr, str):
              weight = edge_attr
              edge_attr = [weight]
          if not (set(edge_attr).issubset(set(input_ddf.columns))):
              raise ValueError(
                  "edge_attr column name not found in input."
                  "Recheck the edge_attr parameter"
              )
          self.properties.weighted = True
  
          if len(edge_attr) == 1:
              input_ddf = input_ddf.rename(
                  columns={edge_attr[0]: self.edgeWeightCol}
              )
              value_col_names = [self.edgeWeightCol]
          elif len(edge_attr) == 3:
              weight_col, id_col, type_col = edge_attr
              input_ddf = input_ddf[
                  ddf_columns + [weight_col, id_col, type_col]
              ]
              input_ddf.columns = ddf_columns + [
                  self.edgeWeightCol,
                  self.edgeIdCol,
                  self.edgeTypeCol,
              ]
              value_col_names = [
                  self.edgeWeightCol,
                  self.edgeIdCol,
                  self.edgeTypeCol,
              ]
          else:
              raise ValueError(
                  "Only 1 or 3 values may be provided" "for edge_attr"
              )
  
          # The symmetrize step may add additional edges with unknown
          # ids and types for an undirected graph.  Therefore, only
          # directed graphs may be used with ids and types.
          # FIXME: Drop the check in symmetrize.py as it is redundant
          if len(edge_attr) == 3:
              if not self.properties.directed:
                  raise ValueError(
                      "User-provided edge ids and/or edge "
                      "types are not permitted for an "
                      "undirected graph."
                  )
  
      else:
          value_col_names = {}
          if weight is not None:
              value_col_names[weight] = self.edgeWeightCol
              self.properties.weighted = True
          if edge_id is not None:
              value_col_names[edge_id] = self.edgeIdCol
          if edge_type is not None:
              value_col_names[edge_type] = self.edgeTypeCol
  
          if len(value_col_names.keys()) > 0:
              input_ddf = input_ddf.rename(columns=value_col_names)
          value_col_names = list(value_col_names.values())
  
      ddf_columns += value_col_names
      input_ddf = input_ddf[ddf_columns]
  
      if len(value_col_names) == 0:
          source_col, dest_col = symmetrize(
              input_ddf,
              source,
              destination,
              multi=self.properties.multi_edge,
              symmetrize=not self.properties.directed,
          )
          value_col = None
      else:
          source_col, dest_col, value_col = symmetrize(
              input_ddf,
              source,
              destination,
              value_col_names,
              multi=self.properties.multi_edge,
              symmetrize=not self.properties.directed,
          )
  
      if isinstance(source_col, dask_cudf.Series):
          # Create a dask_cudf dataframe from the cudf series obtained
          # from symmetrization
          input_ddf = source_col.to_frame()
          input_ddf = input_ddf.rename(
              columns={source_col.name: source}
          )
          input_ddf[destination] = dest_col
      else:
          # Multi column dask_cudf dataframe
          input_ddf = dask_cudf.concat([source_col, dest_col], axis=1)
  
      if value_col is not None:
          for vc in value_col_names:
              input_ddf[vc] = value_col[vc]
  
      self.input_df = input_ddf
  
      #
      # Keep all of the original parameters so we can lazily
      # evaluate this function
      #
  
      # FIXME: Edge Attribute not handled
      # FIXME: the parameter below is no longer used for unrenumbering
      self.properties.renumber = renumber
      self.source_columns = source
      self.destination_columns = destination
      self.weight_column = weight
  
      # If renumbering is not enabled, this function will only create
      # the edgelist_df and not do any renumbering.
      # C++ renumbering is enabled by default for algorithms that
      # support it (but only called if renumbering is on)
      self.compute_renumber_edge_list(
          transposed=store_transposed,
          legacy_renum_only=legacy_renum_only,
      )
  
      if renumber is False:
          self.properties.renumbered = False
          src_col_name = self.source_columns
          dst_col_name = self.destination_columns
  
      else:
          # If 'renumber' is set to 'True', an extra renumbering (python)
          # occurs if there are non-integer or multi-columns vertices
          self.properties.renumbered = self.renumber_map.is_renumbered
  
          src_col_name = self.renumber_map.renumbered_src_col_name
          dst_col_name = self.renumber_map.renumbered_dst_col_name
  
      ddf = self.edgelist.edgelist_df
  
      # Get the edgelist dtypes
      self.vertex_type = ddf[src_col_name].dtype
      if simpleDistributedGraphImpl.edgeWeightCol in ddf.columns:
          self.weight_type = ddf[
              simpleDistributedGraphImpl.edgeWeightCol
          ].dtype
      if simpleDistributedGraphImpl.edgeIdCol in ddf.columns:
          self.edge_id_type = ddf[
              simpleDistributedGraphImpl.edgeIdCol
          ].dtype
      if simpleDistributedGraphImpl.edgeTypeCol in ddf.columns:
          self.edge_type_id_type = ddf[
              simpleDistributedGraphImpl.edgeTypeCol
          ].dtype
  
      graph_props = GraphProperties(
          is_multigraph=self.properties.multi_edge,
          is_symmetric=not self.properties.directed,
      )
      ddf = ddf.repartition(npartitions=len(workers) * 2)
      workers = _client.scheduler_info()["workers"].keys()
      persisted_keys_d = persist_dask_df_equal_parts_per_worker(
          ddf, _client, return_type="dict"
      )
      del ddf
  
      delayed_tasks_d = {
          w: delayed(simpleDistributedGraphImpl._make_plc_graph)(
              Comms.get_session_id(),
              edata,
              graph_props,
              src_col_name,
              dst_col_name,
              store_transposed,
              self.vertex_type,
              self.weight_type,
              self.edge_id_type,
              self.edge_type_id_type,
          )
          for w, edata in persisted_keys_d.items()
      }
      del persisted_keys_d
      self._plc_graph = {
          w: _client.compute(
              delayed_task,
              workers=w,
              allow_other_workers=False,
              pure=False,
          )
          for w, delayed_task in delayed_tasks_d.items()
      }
      del delayed_tasks_d
      run_gc_on_dask_cluster(_client)
      wait(list(self._plc_graph.values()))
      run_gc_on_dask_cluster(_client)
  
imports: import gc
  from typing import Union
  import warnings
  import cudf
  import cupy
  import dask
  import dask_cudf
  from dask import delayed
  from dask.distributed import wait, default_client
  import numpy
  from pylibcugraph import MGGraph, ResourceHandle, GraphProperties, get_two_hop_neighbors, select_random_vertices
  from cugraph.structure import graph_primtypes_wrapper
  from cugraph.structure.graph_primtypes_wrapper import Direction
  from cugraph.structure.number_map import NumberMap
  from cugraph.structure.symmetrize import symmetrize
  from cugraph.dask.common.part_utils import get_persisted_df_worker_map, persist_dask_df_equal_parts_per_worker
  from cugraph.dask.common.mg_utils import run_gc_on_dask_cluster
  from cugraph.dask import get_n_workers
  import cugraph.dask.comms.comms
label_desc: Use image node_id 3 for calling the simpleDistributedGraphImpl obj's underlying member method code with example usage: obj.__from_edgelist(input_ddf, source, destination, edge_attr, weight, edge_id, edge_type, renumber, store_transposed, legacy_renum_only) without return types