code: def ego_graph(input_graph, n, radius=1, center=True):
      """
      Compute the induced subgraph of neighbors centered at node n,
      within a given radius.
  
      Parameters
      ----------
      input_graph : cugraph.Graph
          Graph or matrix object, which should contain the connectivity
          information. Edge weights, if present, should be single or double
          precision floating point values.
  
      n : int, list or cudf Series or Dataframe, dask_cudf Series or DataFrame
          A node or a list or cudf.Series of nodes or a cudf.DataFrame if nodes
          are represented with multiple columns. If a cudf.DataFrame is provided,
          only the first row is taken as the node input.
  
      radius: integer, optional (default=1)
          Include all neighbors of distance<=radius from n.
  
      center: bool, optional
          Defaults to True. False is not supported
  
      Returns
      -------
      ego_edge_lists : dask_cudf.DataFrame
          Distributed GPU data frame containing all induced sources identifiers,
          destination identifiers, edge weights
      seeds_offsets: dask_cudf.Series
          Distributed Series containing the starting offset in the returned edge list
          for each seed.
  
      """
  
      # Initialize dask client
      client = default_client()
  
      if isinstance(n, (int, list)):
          n = cudf.Series(n)
      elif not isinstance(
          n,
          (
              cudf.Series,
              dask_cudf.Series,
              cudf.DataFrame,
              dask_cudf.DataFrame,
          ),
      ):
          raise TypeError(
              f"'n' must be either an integer or a list or a "
              f"cudf or dask_cudf Series or DataFrame, got: {type(n)}"
          )
  
      # n uses "external" vertex IDs, but since the graph has been
      # renumbered, the node ID must also be renumbered.
      if input_graph.renumbered:
          n = input_graph.lookup_internal_vertex_id(n)
          n_type = input_graph.edgelist.edgelist_df.dtypes[0]
      else:
          n_type = input_graph.input_df.dtypes[0]
  
      if isinstance(n, (cudf.Series, cudf.DataFrame)):
          n = dask_cudf.from_cudf(
              n, npartitions=min(input_graph._npartitions, len(n))
          )
      n = n.astype(n_type)
  
      n = persist_dask_df_equal_parts_per_worker(
          n, client, return_type="dict"
      )
      do_expensive_check = False
  
      result = [
          client.submit(
              _call_ego_graph,
              Comms.get_session_id(),
              input_graph._plc_graph[w],
              n_[0] if n_ else cudf.Series(dtype=n_type),
              radius,
              do_expensive_check,
              workers=[w],
              allow_other_workers=False,
          )
          for w, n_ in n.items()
      ]
      wait(result)
  
      cudf_result = [
          client.submit(convert_to_cudf, cp_arrays)
          for cp_arrays in result
      ]
  
      wait(cudf_result)
  
      ddf = dask_cudf.from_delayed(cudf_result).persist()
      wait(ddf)
  
      wait(
          [
              (r.release(), c_r.release())
              for r, c_r in zip(result, cudf_result)
          ]
      )
  
      ddf = ddf.sort_values("labels")
  
      # extract offsets from segmented ego_graph dataframes
      offsets = ddf["labels"].value_counts().compute().sort_index()
      offsets = cudf.concat([cudf.Series(0), offsets])
      offsets = (
          dask_cudf.from_cudf(
              offsets, npartitions=min(input_graph._npartitions, len(n))
          )
          .cumsum()
          .astype(n_type)
      )
  
      ddf = ddf.drop(columns="labels")
  
      if input_graph.renumbered:
          ddf = input_graph.unrenumber(ddf, "src")
          ddf = input_graph.unrenumber(ddf, "dst")
  
      return ddf, offsets
  
file_path: cugraph/python/cugraph/cugraph/dask/community/egonet.py
imports: from dask.distributed import wait, default_client
  import cugraph.dask.comms.comms
  import dask_cudf
  import cudf
  from cugraph.dask.common.part_utils import persist_dask_df_equal_parts_per_worker
  from pylibcugraph import ResourceHandle, ego_graph
label_desc: Use image node_id 4 for calling a global function with example usage: ego_graph(input_graph, n, radius, center) and returns: ddf,  offsets