code: def _train(self):
      log.info("gpu train data generator")
      client_id = os.getpid()
      graph = DistGraphClient(
          self.config,
          self.config.shard_num,
          self.ip_list_file,
          client_id,
      )
      rank = self._worker_info.fid
      nrank = self._worker_info.num_workers
  
      pipeline = []
      pipeline.append(
          NodeGenerator(self.config, graph, rank=rank, nrank=nrank)
      )
      pipeline.append(
          WalkGenerator(self.config, graph, rank=rank, nrank=nrank)
      )
      pipeline.append(
          EgoGraphGenerator(self.config, graph, rank=rank, nrank=nrank)
      )
      pipeline.append(
          PairGenerator(self.config, graph, rank=rank, nrank=nrank)
      )
  
      self.generator = Generator()
      for p in pipeline:
          self.generator = self.generator.apply(p)
  
      cc = 0
      for data in self.generator():
          yield data
          cc += 1
  
imports: import os
  import sys
  import time
  import warnings
  import functools
  import numpy
  from collections import defaultdict, OrderedDict
  import paddle.distributed.fleet
  import pgl
  from pgl.graph_kernel import skip_gram_gen_pair
  from pgl.utils.logger import log
  from pgl.distributed import DistGraphClient, DistGraphServer
  from pgl.utils.data import Dataloader, StreamDataset
  from pgl.distributed import helper
  from pgl.utils.logger import log
  from utils.config import prepare_config
  from datasets.node import NodeGenerator
  from datasets.walk import WalkGenerator
  from datasets.ego_graph import ego_graph_sample, EgoGraphGenerator
  from datasets.pair import PairGenerator
label_desc: Use image node_id 3 for calling the SageDataset obj's underlying member method code with example usage: obj._train() without return types