code: def mapper(args, input_file, lock_list, proc_index):
      """
      mapper
      """
      logger.debug("processing %s" % input_file)
      start_time = time.time()
      cache = []
      for i in range(args.part_num):
          cache.append([])
  
      with open(input_file, "r") as reader:
          for line in reader:
              line = line.rstrip("\r\n")
              fields = line.split("\t")
              if args.node_type_shard:  # node type shard
                  nid = int(fields[1])
                  cache[nid % args.part_num].append(line)
              else:  # edge shard
                  src_nid = int(fields[0])
                  cache[src_nid % args.part_num].append(line)
  
                  if args.symmetry:
                      dst_nid = int(fields[1])
                      if src_nid != dst_nid:
                          cache[dst_nid % args.part_num].append(line)
  
      read_time = time.time()
  
      for i in range(proc_index, args.part_num + proc_index):
          index = i % args.part_num
          lock_list[index].acquire()
          with open(
              os.path.join(args.output_dir, "part-%05d" % index), "a+"
          ) as writer:
              for item in cache[index]:
                  writer.write("%s\n" % item)
          lock_list[index].release()
  
      output_time = time.time()
      logger.debug(
          "processed %s, read_time[%f] write_time[%f]"
          % (
              input_file,
              (read_time - start_time),
              (output_time - read_time),
          )
      )
  
file_path: PGL/apps/PGLBox/sharding_tool/graph_sharding.py
imports: import sys
  import os
  import logging
  import time
  import json
  import shutil
  import argparse
  import multiprocessing
  from multiprocessing import Lock
  from multiprocessing import Process
label_desc: Use image node_id 1 for calling a global function with example usage: mapper(args, input_file, lock_list, proc_index) without return types