code: def sample_graph(
      G,
      label_df,
      output_path,
      seed=42,
      batch_size=500,
      seeds_per_call=400000,
      batches_per_partition=100,
      fanout=[5, 5, 5],
      num_epochs=1,
      train_perc=0.8,
      val_perc=0.5,
      sampling_kwargs={},
  ):
      cupy.random.seed(seed)
      train_df, test_df = label_df.random_split(
          [train_perc, 1 - train_perc], random_state=seed, shuffle=True
      )
      val_df, test_df = label_df.random_split(
          [val_perc, 1 - val_perc], random_state=seed, shuffle=True
      )
  
      total_time = 0.0
      for epoch in range(num_epochs):
          steps = [("train", train_df), ("test", test_df)]
          if epoch == num_epochs - 1:
              steps.append(("val", val_df))
  
          for step, batch_df in steps:
              batch_df = batch_df.sample(frac=1.0, random_state=seed)
  
              if step == "val":
                  output_sample_path = os.path.join(
                      output_path, "val", "samples"
                  )
              else:
                  output_sample_path = os.path.join(
                      output_path,
                      f"epoch={epoch}",
                      f"{step}",
                      "samples",
                  )
              os.makedirs(output_sample_path)
  
              sampler = BulkSampler(
                  batch_size=batch_size,
                  output_path=output_sample_path,
                  graph=G,
                  fanout_vals=fanout,
                  with_replacement=False,
                  random_state=seed,
                  seeds_per_call=seeds_per_call,
                  batches_per_partition=batches_per_partition,
                  log_level=logging.INFO,
                  **sampling_kwargs,
              )
  
              n_workers = len(
                  default_client().scheduler_info()["workers"]
              )
  
              meta = cudf.DataFrame(
                  {
                      "node": cudf.Series(dtype="int64"),
                      "batch": cudf.Series(dtype="int32"),
                  }
              )
  
              batch_df = batch_df.map_partitions(
                  _make_batch_ids, batch_size, n_workers, meta=meta
              )
  
              # should always persist the batch dataframe or performance may be suboptimal
              batch_df = batch_df.persist()
  
              print("created batches")
  
              start_time = perf_counter()
              sampler.add_batches(
                  batch_df,
                  start_col_name="node",
                  batch_col_name="batch",
              )
              sampler.flush()
              end_time = perf_counter()
              print("flushed all batches")
              total_time += end_time - start_time
  
      return total_time
  
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 6 for calling a global function with example usage: sample_graph(G, label_df, output_path, seed, batch_size, seeds_per_call, batches_per_partition, fanout, num_epochs, train_perc, val_perc, sampling_kwargs) and returns: total_time