您可以使用级联文档中提到的拆分模式。举个例子：

public static void main(String[] args) {
    // source and sink
    Scheme sourceScheme = new TextLine(new Fields("line"));
    Tap source = new FileTap(sourceScheme, args[0]);
    Fields sinkFields = new Fields("word", "count");
    Scheme sinkScheme = new TextLine(sinkFields, sinkFields);
    Tap sink_one = new FileTap(sinkScheme, "out-one.txt");
    Tap sink_two = new FileTap(sinkScheme, "out-two.txt");
    // the pipe assembly
    Pipe assembly = new Pipe("wordcount");
    String regex = "\\w+";
    Function function = new RegexGenerator(new Fields("word"), regex);
    assembly = new Each(assembly, new Fields("line"), function);
    Aggregator count = new Count(new Fields("count"));
    // ...split into two pipes
    Pipe countOne = new Pipe("count-one", assembly);
    countOne = new GroupBy(countOne, new Fields("word"));
    countOne = new Every(countOne, count);
    Pipe countTwo = new Pipe("count-two", assembly);
    countTwo = new GroupBy(countTwo, new Fields("word"));
    countTwo = new Every(countTwo, count);
    // create the flow
    final List<Pipe> pipes = new ArrayList<Pipe>(2);
    pipes.add(countOne);
    pipes.add(countTwo);
    final Map<String, Tap> sinks = new HashMap<String, Tap>();
    sinks.put("count-one", sink_one);
    sinks.put("count-two", sink_two);
    FlowConnector flowConnector = new LocalFlowConnector();
    Flow flow = flowConnector.connect(source, sinks, pipes);
    flow.complete();
}

拆分模式在级联用户指南中：http://docs.cascading.org/cascading/2.1/userguide/htmlsingle/#n21362
另一个（更简单的）例子包含在“不耐烦者的级联”第5部分和第6部分：
https://github.com/cascading/impatient/wiki/part-5
https://github.com/cascading/impatient/wiki/part-6
关于上面显示的代码，有一点是它似乎缺少 report1Pipe 以及 report2Pipe . 要使用拆分模式，每个分支都需要一个名称，并且名称必须不同。
引发异常的原因是有两个分支都继承了管道程序集中早期版本的相同名称。例如，这些 flowDef.addSink(..) 调用对flow planner不明确。
所以在“不耐烦”的第5部分中，看看“d”、“df”和“tf”分支是如何在操作中命名的。
级联要求这种命名似乎有点违反直觉，但在大型复杂工作流中，当您将故障陷阱、调试等附加到特定分支时，这种命名就变得非常重要。
或者，clojure中的cascalog dsl更具声明性，因此它直接由语言处理——分支是子查询，陷阱等在子查询的闭包中处理。