Loop-in on GroupBy - Агрегация в каскадировании

Question

Когда агрегирование в GroupBy применяется на любой трубке, в полученных полях есть группа группировки &. Я хочу все поля в дополнение к агрегированной стоимости в Каскадные

Пример: вход:

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

Нормальный Агрегированный Выход:

A|17 
B|10 

Я хочу: выход же в качестве входных данных, но в дополнении агрегированных колонок, как well--

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); 

Answer 1

Лучший способом я нашел, чтобы создать собственный Buffer и использовать список, чтобы сохранить входящий TupleEntry вычислить сумму и записать сумму перебирая предыдущего сохраненного список

@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);