@Test
public void testCoGroup() throws Exception {

  Pump left = Pump.prime("left")
      .each(new RegexFilter("^[0-9]+$", false), "line")
      .retain("line")
      .coerce("line", int.class)
      .each(new DateFormatter(new Fields("date"), "yyyy-MM-dd"))
      .retain("date")
      .groupby("date")
      .every(new Count(new Fields("count")));
  Pump right = Pump.prime("right")
      .each(new RegexSplitter(new Fields("date", "tag"),"\t"), "line");
  Pipe pipe = Pump.cogroup(left, right, "date")
      .retain("date", "count", "tag")
      .toPipe();
  Map<String, Tap> inputTaps = new HashMap<String, Tap>() {{
    put("left", Tests.getInTap());
    put("right", Tests.getTap(INPUT2_PATH));
  }};
  CascadingHelper.get().getFlowConnector().connect(inputTaps, Tests.getOutTap(), pipe).complete();
  assertEquals(Arrays.asList("1970-01-01\t1\tfirst", "1970-01-02\t2\tsecond"), getOutputStrings());
}