title: Why does this Pig UDF Result in an "Error: Java heap space" Given that I am Spilling the DataBag to Disk?
body: <p>Here is my UDF:</p>
  <pre><code>public DataBag exec(Tuple input) throws IOException { 
      Aggregate aggregatedOutput = null;
      
      int spillCount = 0;
  
      DataBag outputBag = BagFactory.newDefaultBag(); 
      DataBag values = (DataBag)input.get(0);
      for (Iterator&lt;Tuple&gt; iterator = values.iterator(); iterator.hasNext();) {
          Tuple tuple = iterator.next();
          //spillCount++;
          ...
          if (some condition regarding current input tuple){
              //do something to aggregatedOutput with information from input tuple
          } else {
              //Because input tuple does not apply to current aggregateOutput
              //return current aggregateOutput and apply input tuple
              //to new aggregateOutput
              Tuple returnTuple = aggregatedOutput.getTuple();
              outputBag.add(returnTuple);
              spillCount++;
              aggregatedOutputTuple = new Aggregate(tuple);
              
              
              if (spillCount == 1000) {
                  outputBag.spill();
                  spillCount = 0;
              }
          }
      }
      return outputBag; 
  }
  </code></pre>
  <p>Please focus on the fact that for every 1000 input tuples, the bag spills to disk. I have set this number as low as 50 and as high as 100,000 yet still receive the memory error:</p>
  <pre><code>Pig logfile dump:
  
  Backend error message
  ---------------------
  Error: Java heap space
  
  Pig Stack Trace
  ---------------
  ERROR 2997: Unable to recreate exception from backed error: Error: Java heap space
  </code></pre>
  <p>What can I do to solve this? It is processing about a million rows.</p>
  <h2>HERE IS THE SOLUTION</h2>
  <p>Using the Accumulator interface:</p>
  <pre><code>public class Foo extends EvalFunc&lt;DataBag&gt; implements Accumulator&lt;DataBag&gt; {
      private DataBag outputBag = null;
      private UltraAggregation currentAggregation = null;
      
      public void accumulate(Tuple input) throws IOException {
          DataBag values = (DataBag)input.get(0);
          Aggregate aggregatedOutput = null;
          outputBag = BagFactory.getInstance().newDefaultBag();
          
          for (Iterator&lt;Tuple&gt; iterator = values.iterator(); iterator.hasNext();) {
              Tuple tuple = iterator.next();
              ...
              if (some condition regarding current input tuple){
                  //do something to aggregatedOutput with information from input tuple
              } else {
                  //Because input tuple does not apply to current aggregateOutput
                  //return current aggregateOutput and apply input tuple
                  //to new aggregateOutput
                  outputBag.add(aggregatedOutput.getTuple());
                  aggregatedOutputTuple = new Aggregate(tuple);
              }
          }
      }
      
      // Called when all tuples from current key have been passed to accumulate
      public DataBag getValue() {
          //Add final current aggregation
          outputBag.add(currentAggregation.getTuple());
          return outputBag;
      }
      // This is called after getValue()
      // Not sure if these commands are necessary as they are repeated in beginning of accumulate
      public void cleanup() {
          outputBag = null;
          currentAggregation = null;
      }
      
      public DataBag exec(Tuple input) throws IOException {
          // Same as above ^^ but this doesn't appear to ever be called.
      }
      
      public Schema outputSchema(Schema input) {
          try {
              return new Schema(new FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(), input), bagSchema, DataType.BAG));
          } catch {FrontendException e) {
              e.printStackTrace();
              return null;
          }
      }
      
      class Aggregate {
          ...
          public Tuple getTuple() {
              Tuple output = TupleFactory.getInstance().newTuple(OUTPUT_TUPLE_SIZE);
              try {
                  output.set(0, val);
                  ...
              } catch (ExecException e) {
                  e.printStackTrace();
                  return null;
              }
          }
          ...
      }
  }
  </code></pre>