Pump { public AggregatorPump every(Aggregator agg, String... args) { return new AggregatorPump(this, agg, args); }  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 testEvery() throws IOException { String inputPath = "/tmp/TestPump/group_by_sec_sort"; FileSystem.get(new Configuration()).delete(new Path(inputPath), true); Tap inTap = new Hfs(new SequenceFile(new Fields("key1", "key2")), inputPath); TupleEntryCollector collector = inTap.openForWrite(new HadoopFlowProcess()); collector.add(new Tuple("key1", "value1")); collector.add(new Tuple("key1", "value2")); collector.close(); Pump pump = Pump.prime() .retain("key1", "key2") .groupby("key1") .every(new First()); Pipe tail = pump.toPipe(); FlowDef flowDef = new FlowDef() .addSource("input", inTap) .addTail(tail) .addSink(tail, new Hfs(new TextLine(new Fields("key1")), Tests.OUTPUT_PATH)); CascadingHelper.get().getFlowConnector().connect(flowDef).complete(); List<String> outputStrings = getOutputStrings(); assertEquals(Arrays.asList("key1\tvalue1"), outputStrings); }