@Override
  public String exec(Tuple t) throws IOException {
    return SummaryData.toPrettyJSON(sumUp(getInputSchema(), t));
  }
@Test
  public void testAlgebraic() throws IOException {
    Summary.Initial initial = new Summary.Initial();
    Summary.Intermediate intermediate1 = new Summary.Intermediate();
    Summary.Intermediate intermediate2 = new Summary.Intermediate();
    Summary.Final finall = new Summary.Final();

    DataBag combinedRedIn = bf.newDefaultBag();
    for (int r = 0; r < 5; r++) {
      DataBag combinedMapOut = bf.newDefaultBag();
      for (int m = 0; m < 5; m++) {
        DataBag mapOut = bf.newDefaultBag();
        for (Tuple t : TEST_BAG) {
          Tuple exec = initial.exec(t(b(t)));
          mapOut.add(exec);
        }
        Tuple exec = intermediate1.exec(t(mapOut));
        validate((String) exec.get(0), 1);
        combinedMapOut.add(exec);
      }
      combinedRedIn.add(intermediate2.exec(t(combinedMapOut)));
    }
    String result = finall.exec(t(combinedRedIn));
    validate(result, 5 * 5);
  }