code: def generate_rmat_dataset(
      dataset,
      seed=62,
      labeled_percentage=0.01,
      num_labels=256,
      reverse_edges=False,
      add_edge_types=False,
  ):
      """
      Generates an rmat dataset.  Currently does not support heterogeneous datasets.
  
      Parameters
      ----------
      dataset: The specifier of the rmat dataset (i.e. rmat_20_16)
      seed: The seed to use for random number generation
      num_labels: The number of classes for the labeled nodes
      reverse_edges: Whether to reverse the edges in the edgelist (should be True for DGL, False, for PyG)
      """
  
      dataset = dataset.split("_")
      scale = int(dataset[1])
      edgefactor = int(dataset[2])
  
      dask_edgelist_df = generate_edgelist_rmat(
          scale=scale,
          edgefactor=edgefactor,
          seed=seed,
          unweighted=True,
          mg=True,
      )
      dask_edgelist_df = dask_edgelist_df.astype("int64")
      dask_edgelist_df = dask_edgelist_df.reset_index(drop=True)
  
      dask_edgelist_df = renumber_ddf(dask_edgelist_df).persist()
  
      dask_edgelist_df = symmetrize_ddf(dask_edgelist_df).persist()
  
      if add_edge_types:
          dask_edgelist_df["etp"] = cupy.int32(
              0
          )  # doesn't matter what the value is, really
  
      # generator = np.random.default_rng(seed=seed)
      num_labeled_nodes = int(2 ** (scale + 1) * labeled_percentage)
      label_df = pd.DataFrame(
          {
              "node": np.arange(num_labeled_nodes),
              # 'label': generator.integers(0, num_labels - 1, num_labeled_nodes).astype('float32')
          }
      )
  
      n_workers = len(default_client().scheduler_info()["workers"])
      dask_label_df = ddf.from_pandas(
          label_df, npartitions=n_workers * 2
      )
      del label_df
      gc.collect()
  
      dask_label_df = dask_cudf.from_dask_dataframe(dask_label_df)
  
      node_offsets = {"paper": 0}
      edge_offsets = {("paper", "cites", "paper"): 0}
      total_num_nodes = int(
          dask_cudf.concat([dask_edgelist_df.src, dask_edgelist_df.dst])
          .nunique()
          .compute()
      )
  
      if reverse_edges:
          dask_edgelist_df = dask_edgelist_df.rename(
              columns={"src": "dst", "dst": "src"}
          )
  
      return (
          dask_edgelist_df,
          dask_label_df,
          node_offsets,
          edge_offsets,
          total_num_nodes,
      )
  
imports: import logging
  import warnings
  import argparse
  import traceback
  from cugraph.testing.mg_utils import generate_edgelist_rmat, get_allocation_counts_dask_lazy, sizeof_fmt, get_peak_output_ratio_across_workers, start_dask_client, stop_dask_client, enable_spilling
  from cugraph.structure.symmetrize import symmetrize
  from cugraph.gnn import BulkSampler
  import cugraph
  import json
  import re
  import os
  import gc
  from time import sleep, perf_counter
  from math import ceil
  import pandas
  import numpy
  import cupy
  import cudf
  import dask_cudf
  import dask.dataframe
  from dask.distributed import default_client
  from cugraph.dask import get_n_workers
  from typing import Optional, Union, Dict
label_desc: Use image node_id 8 for calling a global function with example usage: generate_rmat_dataset(dataset, seed, labeled_percentage, num_labels, reverse_edges, add_edge_types) and returns: dask_edgelist_df,  dask_label_df,  node_offsets,  edge_offsets,  total_num_nodes