docstr: `batch_size` is the # of Dask graphs sent to the cluster
  simultaneously for processing.
  
  One element in the batch represents 1 Dask graph.
  The Dask graph involves reading 30 arrays (one is 1.44GB)
  and concatenating them into a Dask array.
  Then, it does FFT computations across chunks of the Dask array.
  It saves the FFT-ed version of the Dask array as an output file.
  
  If `trigger_object_spill` is True, then we send work to
  the cluster such that each worker gets the number of graphs
  that would exceed the worker memory, triggering object spills.
  We use the estimated peak memory consumption to determine
  how many graphs should be sent.
  
  If `error_rate` is True, we throw an exception at the Data
  load layer as per error rate.
docstr_tok: ["`","batch_size","`","is","the","#","of","Dask","graphs","sent","to","the","cluster","simultaneously","for","processing",".","One","element","in","the","batch","represents","1","Dask","graph",".","The","Dask","graph","involves","reading","30","arrays","(","one","is","1.44GB",")","and","concatenating","them","into","a","Dask","array",".","Then",",","it","does","FFT","computations","across","chunks","of","the","Dask","array",".","It","saves","the","FFT-ed","version","of","the","Dask","array","as","an","output","file",".","If","`","trigger_object_spill","`","is","True",",","then","we","send","work","to","the","cluster","such","that","each","worker","gets","the","number","of","graphs","that","would","exceed","the","worker","memory",",","triggering","object","spills",".","We","use","the","estimated","peak","memory","consumption","to","determine","how","many","graphs","should","be","sent",".","If","`","error_rate","`","is","True",",","we","throw","an","exception","at","the","Data","load","layer","as","per","error","rate","."]
code: def __init__(
      self,
      num_workers: int,
      worker_obj_store_size_in_gb: int,
      trigger_object_spill: bool,
      error_rate: float,
  ):
      """
      `batch_size` is the # of Dask graphs sent to the cluster
      simultaneously for processing.
  
      One element in the batch represents 1 Dask graph.
      The Dask graph involves reading 30 arrays (one is 1.44GB)
      and concatenating them into a Dask array.
      Then, it does FFT computations across chunks of the Dask array.
      It saves the FFT-ed version of the Dask array as an output file.
  
      If `trigger_object_spill` is True, then we send work to
      the cluster such that each worker gets the number of graphs
      that would exceed the worker memory, triggering object spills.
      We use the estimated peak memory consumption to determine
      how many graphs should be sent.
  
      If `error_rate` is True, we throw an exception at the Data
      load layer as per error rate.
      """
      self.error_rate = error_rate
  
      if trigger_object_spill:
          num_graphs_per_worker = (
              int(
                  math.floor(
                      worker_obj_store_size_in_gb
                      / PEAK_MEMORY_CONSUMPTION_IN_GB
                  )
              )
              + 1
          )
      else:
          num_graphs_per_worker = int(
              math.floor(
                  worker_obj_store_size_in_gb
                  / PEAK_MEMORY_CONSUMPTION_IN_GB
              )
          )
      self.batch_size = num_graphs_per_worker * num_workers
  
imports: import argparse
  import json
  import logging
  import math
  import os
  import random
  import time
  from copy import copy, deepcopy
  from typing import List, Tuple
  import dask.array
  import numpy
  import xarray
  import ray
  from ray._private.test_utils import monitor_memory_usage
  from ray.util.dask import ray_dask_get
label_desc: Use image node_id 1 to create a new TestSpec object  with example: obj = TestSpec(num_workers, worker_obj_store_size_in_gb, trigger_object_spill, error_rate)