code: def gen_node_data(
      rank, world_size, num_parts, id_lookup, ntid_ntype_map, schema_map
  ):
      """
      For this data processing pipeline, reading node files is not needed. All the needed information about
      the nodes can be found in the metadata json file. This function generates the nodes owned by a given
      process, using metis partitions.
  
      Parameters:
      -----------
      rank : int
          rank of the process
      world_size : int
          total no. of processes
      num_parts : int
          total no. of partitions
      id_lookup : instance of class DistLookupService
         Distributed lookup service used to map global-nids to respective partition-ids and
         shuffle-global-nids
      ntid_ntype_map :
          a dictionary where keys are node_type ids(integers) and values are node_type names(strings).
      schema_map:
          dictionary formed by reading the input metadata json file for the input dataset.
  
          Please note that, it is assumed that for the input graph files, the nodes of a particular node-type are
          split into `p` files (because of `p` partitions to be generated). On a similar node, edges of a particular
          edge-type are split into `p` files as well.
  
          #assuming m nodetypes present in the input graph
          "num_nodes_per_chunk" : [
              [a0, a1, a2, ... a<p-1>],
              [b0, b1, b2, ... b<p-1>],
              ...
              [m0, m1, m2, ... m<p-1>]
          ]
          Here, each sub-list, corresponding a nodetype in the input graph, has `p` elements. For instance [a0, a1, ... a<p-1>]
          where each element represents the number of nodes which are to be processed by a process during distributed partitioning.
  
          In addition to the above key-value pair for the nodes in the graph, the node-features are captured in the
          "node_data" key-value pair. In this dictionary the keys will be nodetype names and value will be a dictionary which
          is used to capture all the features present for that particular node-type. This is shown in the following example:
  
          "node_data" : {
              "paper": {       # node type
                  "feat": {   # feature key
                      "format": {"name": "numpy"},
                      "data": ["node_data/paper-feat-part1.npy", "node_data/paper-feat-part2.npy"]
                  },
                  "label": {   # feature key
                      "format": {"name": "numpy"},
                      "data": ["node_data/paper-label-part1.npy", "node_data/paper-label-part2.npy"]
                  },
                  "year": {   # feature key
                      "format": {"name": "numpy"},
                      "data": ["node_data/paper-year-part1.npy", "node_data/paper-year-part2.npy"]
                  }
              }
          }
          In the above textual description we have a node-type, which is paper, and it has 3 features namely feat, label and year.
          Each feature has `p` files whose location in the filesystem is the list for the key "data" and "foramt" is used to
          describe storage format.
  
      Returns:
      --------
      dictionary :
          dictionary where keys are column names and values are numpy arrays, these arrays are generated by
          using information present in the metadata json file
  
      """
      local_node_data = {}
      for local_part_id in range(num_parts // world_size):
          local_node_data[
              constants.GLOBAL_NID + "/" + str(local_part_id)
          ] = []
          local_node_data[
              constants.NTYPE_ID + "/" + str(local_part_id)
          ] = []
          local_node_data[
              constants.GLOBAL_TYPE_NID + "/" + str(local_part_id)
          ] = []
  
      # Note that `get_idranges` always returns two dictionaries. Keys in these
      # dictionaries are type names for nodes and edges and values are
      # `num_parts` number of tuples indicating the range of type-ids in first
      # dictionary and range of global-nids in the second dictionary.
      type_nid_dict, global_nid_dict = get_idranges(
          schema_map[constants.STR_NODE_TYPE],
          get_ntype_counts_map(
              schema_map[constants.STR_NODE_TYPE],
              schema_map[constants.STR_NUM_NODES_PER_TYPE],
          ),
          num_chunks=num_parts,
      )
  
      for ntype_id, ntype_name in ntid_ntype_map.items():
          # No. of nodes in each process can differ significantly in lopsided distributions
          # Synchronize on a per ntype basis
          dist.barrier()
  
          type_start, type_end = (
              type_nid_dict[ntype_name][0][0],
              type_nid_dict[ntype_name][-1][1],
          )
          gnid_start, gnid_end = (
              global_nid_dict[ntype_name][0, 0],
              global_nid_dict[ntype_name][0, 1],
          )
  
          node_partid_slice = id_lookup.get_partition_ids(
              np.arange(gnid_start, gnid_end, dtype=np.int64)
          )  # exclusive
  
          for local_part_id in range(num_parts // world_size):
              cond = node_partid_slice == (
                  rank + local_part_id * world_size
              )
              own_gnids = np.arange(
                  gnid_start, gnid_end, dtype=np.int64
              )
              own_gnids = own_gnids[cond]
  
              own_tnids = np.arange(
                  type_start, type_end, dtype=np.int64
              )
              own_tnids = own_tnids[cond]
  
              local_node_data[
                  constants.NTYPE_ID + "/" + str(local_part_id)
              ].append(
                  np.ones(own_gnids.shape, dtype=np.int64) * ntype_id
              )
              local_node_data[
                  constants.GLOBAL_NID + "/" + str(local_part_id)
              ].append(own_gnids)
              local_node_data[
                  constants.GLOBAL_TYPE_NID + "/" + str(local_part_id)
              ].append(own_tnids)
  
      for k in local_node_data.keys():
          local_node_data[k] = np.concatenate(local_node_data[k])
  
      return local_node_data
  
imports: import gc
  import logging
  import math
  import os
  import sys
  from datetime import timedelta
  from timeit import default_timer
  import constants
  import dgl
  import numpy
  import torch
  import torch.distributed
  import torch.multiprocessing
  from convert_partition import create_dgl_object, create_metadata_json
  from dataset_utils import get_dataset
  from dist_lookup import DistLookupService
  from globalids import assign_shuffle_global_nids_edges, assign_shuffle_global_nids_nodes, lookup_shuffle_global_nids_edges
  from gloo_wrapper import allgather_sizes, alltoallv_cpu, gather_metadata_json
  from utils import augment_edge_data, DATA_TYPE_ID, get_edge_types, get_etype_featnames, get_gid_offsets, get_gnid_range_map, get_idranges, get_node_types, get_ntype_counts_map, get_ntype_featnames, map_partid_rank, memory_snapshot, read_json, read_ntype_partition_files, REV_DATA_TYPE_ID, write_dgl_objects, write_metadata_json
label_desc: Use image node_id 1 for calling a global function with example usage: gen_node_data(rank, world_size, num_parts, id_lookup, ntid_ntype_map, schema_map) and returns: local_node_data