GroupBy 上的循环 - 级联中的聚合答案

【问题标题】：Loop-in on GroupBy - Aggregation in CascadingGroupBy 上的循环 - 级联中的聚合
【发布时间】：2016-02-20 21:32:27
【问题描述】：

当在任何管道上应用 GroupBy 聚合时，结果字段具有分组键和聚合值。除了 Cascading

中的聚合值之外，我还想要所有字段

示例：输入：

       A|a1|b1|12
       B|b2|c2|10
       A|a3|a3|5

普通聚合输出：

A|17
B|10

我想要：输出与输入相同，但还有聚合列--

A|a1|b1|12|17
B|b2|c2|10
A|a3|a3|5|17

那么，如何循环 GroupBY 的结果以聚合和打印级联中的所有行？

    Pipe group = new GroupBy(someTuplePipe, groupKey, true);
    group  = new Every(g1, Fields.ALL, new Count(new Fields("sum")), Fields.ALL);

【问题讨论】：

标签： cascading

【解决方案1】：

我发现最好的方法是创建一个自定义 Buffer 并使用 List 来保存传入的 TupleEntry 计算总和并通过迭代之前保存的 List 写出总和

@Override
public void operate(FlowProcess flowProcess, BufferCall bufferCall) {
    Iterator<TupleEntry> iterator = bufferCall.getArgumentsIterator();
    List<TupleEntry> tupleEntryMap = new ArrayList<TupleEntry>();

    int sum = 0;
    while (iterator.hasNext()) {
        TupleEntry inputEntry = new TupleEntry(iterator.next());
        sum += inputEntry.getInteger("amount_field");
        tupleEntryMap.add(inputEntry);
    }

    for (TupleEntry result : tupleEntryMap) {
        result.setInteger("amount_field", sum);
        bufferCall.getOutputCollector().add(result);
    }
}

并按照你上面提到的那样使用它。

Pipe group = new GroupBy(someTuplePipe, groupKey, true);
group  = new Every(group, new CustomBuffer(), Fields.REPLACE);

【讨论】：

那不是和输入流分组然后加入分组流一样吗？
@ganeshtripathi 请详细说明一下。我很难理解你。
@ganeshtripathi 终于明白你想说什么了：D。是的，输出将与您提到的相同，但它会为流程添加至少 2 个额外步骤。
我在谈论更多关于内存和处理方面的内容。
这两个都将采用相同的处理，这就是我所要做的。