code: def test_dask_mg_bfs(dask_client, directed):
      input_data_path = (
          RAPIDS_DATASET_ROOT_DIR_PATH / "netscience.csv"
      ).as_posix()
  
      print(f"dataset={input_data_path}")
      chunksize = dcg.get_chunksize(input_data_path)
  
      ddf = dask_cudf.read_csv(
          input_data_path,
          chunksize=chunksize,
          delimiter=" ",
          names=["src", "dst", "value"],
          dtype=["int32", "int32", "float32"],
      )
  
      def modify_dataset(df):
          temp_df = cudf.DataFrame()
          temp_df["src"] = df["src"] + 1000
          temp_df["dst"] = df["dst"] + 1000
          temp_df["value"] = df["value"]
          return cudf.concat([df, temp_df])
  
      meta = ddf._meta
      ddf = ddf.map_partitions(
          modify_dataset,
          meta=meta,
          token="custom-" + str(random.random()),
      )
  
      df = cudf.read_csv(
          input_data_path,
          delimiter=" ",
          names=["src", "dst", "value"],
          dtype=["int32", "int32", "float32"],
      )
  
      df = modify_dataset(df)
  
      g = cugraph.Graph(directed=directed)
      g.from_cudf_edgelist(df, "src", "dst")
  
      dg = cugraph.Graph(directed=directed)
      dg.from_dask_cudf_edgelist(ddf, "src", "dst")
  
      expected_dist = cugraph.bfs(g, [0, 1000])
      result_dist = dcg.bfs(dg, [0, 1000])
      result_dist = result_dist.compute()
  
      compare_dist = expected_dist.merge(
          result_dist, on="vertex", suffixes=["_local", "_dask"]
      )
  
      err = 0
  
      for i in range(len(compare_dist)):
          if (
              compare_dist["distance_local"].iloc[i]
              != compare_dist["distance_dask"].iloc[i]
          ):
              err = err + 1
      assert err == 0
  
file_path: cugraph/python/cugraph/cugraph/tests/traversal/test_bfs_mg.py
imports: import gc
  import random
  import pytest
  import cudf
  import cugraph
  import dask_cudf
  import cugraph.dask
  from cugraph.testing.utils import RAPIDS_DATASET_ROOT_DIR_PATH
label_desc: Use image node_id 2 for calling a global function with example usage: test_dask_mg_bfs(dask_client, directed) and returns: cudf