code: def put_to_plasma(address):
      def f(index, iterator):
          import pyarrow.plasma as plasma
  
          client = plasma.connect(address)
          part_size = (
              1000000  # TODO: Make subpartition size configurable?
          )
          buffer = []
          sub_index = 0
          for record in iterator:
              if len(buffer) == part_size:
                  res_buffer = process_records(buffer)
                  object_id = client.put(res_buffer)
                  buffer = [record]
                  yield index, sub_index, part_size, object_id, get_node_ip()
                  sub_index += 1
              else:
                  buffer.append(record)
          remain_size = len(buffer)
          if remain_size > 0:
              res_buffer = process_records(buffer)
              object_id = client.put(res_buffer)
              buffer = []
              client.disconnect()
              yield index, sub_index, remain_size, object_id, get_node_ip()
          else:
              client.disconnect()
  
      return f
  
imports: import os
  import types
  import subprocess
  import cloudpickle
  from pyspark.sql import DataFrame
  from torch.utils.data import Dataset, DataLoader
  from zoo.util.utils import get_node_ip
  from zoo.orca.learn.mpi.mpi_runner import MPIRunner
  from zoo.orca.learn.mpi.utils import *
label_desc: Use image node_id 2 for calling a global function with example usage: put_to_plasma(address) and returns: f