datafu-hourglass/src/main/java/datafu/hourglass/mapreduce/CollapsingCombiner.java [203:240]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  public void setOutputDateRange(DateRange dateRange)
  {
    _beginTime = dateRange.getBeginDate().getTime();
    _endTime = dateRange.getEndDate().getTime();
  }
  
  /**
   * Clone a {@link Accumulator} by serializing and deserializing it.
   * 
   * @param acc The accumulator to clone
   * @return The clone accumulator
   */
  private Accumulator<GenericRecord,GenericRecord> cloneAccumulator(Accumulator<GenericRecord,GenericRecord> acc)
  {
    try
    {
      // clone by serializing
      ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
      ObjectOutputStream objStream;
      objStream = new ObjectOutputStream(outputStream);    
      objStream.writeObject(acc);
      objStream.close();
      outputStream.close();
      ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
      ObjectInputStream objInputStream = new ObjectInputStream(inputStream);
      @SuppressWarnings("unchecked")
      Accumulator<GenericRecord,GenericRecord> result = (Accumulator<GenericRecord,GenericRecord>)objInputStream.readObject();
      objInputStream.close();
      inputStream.close();
      return result;
    }
    catch (IOException e)
    {
      throw new RuntimeException(e);
    }
    catch (ClassNotFoundException e)
    {
      throw new RuntimeException(e);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



datafu-hourglass/src/main/java/datafu/hourglass/mapreduce/CollapsingReducer.java [278:315]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  public void setOutputDateRange(DateRange dateRange)
  {
    _beginTime = dateRange.getBeginDate().getTime();
    _endTime = dateRange.getEndDate().getTime();
  }
  
  /**
   * Clone a {@link Accumulator} by serializing and deserializing it.
   * 
   * @param acc The accumulator to clone
   * @return The clone accumulator
   */
  private Accumulator<GenericRecord,GenericRecord> cloneAccumulator(Accumulator<GenericRecord,GenericRecord> acc)
  {
    try
    {
      // clone by serializing
      ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
      ObjectOutputStream objStream;
      objStream = new ObjectOutputStream(outputStream);    
      objStream.writeObject(acc);
      objStream.close();
      outputStream.close();
      ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
      ObjectInputStream objInputStream = new ObjectInputStream(inputStream);
      @SuppressWarnings("unchecked")
      Accumulator<GenericRecord,GenericRecord> result = (Accumulator<GenericRecord,GenericRecord>)objInputStream.readObject();
      objInputStream.close();
      inputStream.close();
      return result;
    }
    catch (IOException e)
    {
      throw new RuntimeException(e);
    }
    catch (ClassNotFoundException e)
    {
      throw new RuntimeException(e);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -