tc.close();
Tap sink = new Hfs(new SequenceFile(new Fields("key", "count")), OUTPUT_PATH);
Pipe pipe = new Pipe("pipe");
pipe = Combiner.assembly(pipe, new CountExactAggregator(), new Fields("key"), new Fields(), new Fields("count"));
CascadingUtil.get().getFlowConnector().connect(source, sink, pipe).complete();
List<Tuple> tuples = getAllTuples(sink);