Pump { public CoGroupPump cogroup(Pump other, String... cogroupFields) { return cogroup(this, other, cogroupFields); }  Set<Class> getEmittedClasses(); final Pipe toPipe(); static Pump prime(); static Pump prime(String pipeName); static Pump prime(Pipe pipe); CoGroupPump cogroup(Pump other, String... cogroupFields); CoGroupPump cogroup(Pump other, Joiner joiner, String... cogroupFields); static CoGroupPump cogroup(Pump left, Pump right, String... cogroupFields); static CoGroupPump cogroup(Pump left, Pump right, Joiner joiner, String... cogroupFields); Pump each(Function function, String... args); Pump each(Filter filter, String... args); Pump failOnNull(String errorText, String... args); Pump unique(String... uniqueFields); GroupByPump groupby(String... fields); Pump aggregateby(AggregateBy.Functor functor, Aggregator aggregator, String... args); Pump average(String valueField, String averageField); Pump count(String countField); Pump first(String... firstFields); Pump sum(String valueField, String sumField); AggregatorPump every(Aggregator agg, String... args); Pump every(Buffer buffer, String... args); Pump retain(String ... fieldsToKeep); Pump discard(String ... fieldsToDiscard); Pump coerce(String field, Class toClass); Pump coerce(Class toClass, String... fieldsToCoerce); Pump coerce(String[] fields, Class<?>[] classes); Pump rename(String field, String toName); Pump replace(String field, String toName); Pump getOrElse(String fromField, String toField, Tuple value); Pump branch();  }
@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()); }