我有以下问题,我试图通过级联解决:我有结构为 o,a,f,i,c 的记录的 csv 文件
我需要按 o、a、f 聚合记录,并对每组 i 和 c 求和。
例如:
100,200,300,5,1
100,200,300,6,2
101,201,301,20,5
101,201,301,21,6
应该产生:
100,200,300,11,3
101,201,301,41,11
我不明白如何合并我拥有的 2 个 Every 实例(我可以同时聚合两个字段吗?)。
你有什么想法吗?
尤西
public class CascMain {
public static void main(String[] args){
Scheme sourceScheme = new TextLine(new Fields("line"));
Tap source = new Lfs(sourceScheme, "/tmp/casc/group.csv");
Scheme sinkScheme = new TextDelimited(new Fields("o", "a", "f", "ti", "tc"), ",");
Tap sink = new Lfs(sinkScheme, "/tmp/casc/output/", SinkMode.REPLACE);
Pipe assembly = new Pipe("agg-pipe");
Function function = new RegexSplitter(new Fields("o", "a", "f", "i", "c"), ",");
assembly = new Each(assembly, new Fields("line"), function);
Pipe groupAssembly = new GroupBy("group", assembly, new Fields("o", "a", "f"));
Sum impSum = new Sum(new Fields("ti"));
Pipe i = new Every(groupAssembly, new Fields("i"), impSum);
Sum clickSum = new Sum(new Fields("tc"));
Pipe c = new Every(groupAssembly, new Fields("c"), clickSum);
// WHAT SHOULD I DO HERE
Properties properties = new Properties();
FlowConnector.setApplicationJarClass(properties, CascMain.class);
FlowConnector flowConnector = new FlowConnector(properties);
Flow flow = flowConnector.connect("agg", source, sink, assembly);
flow.complete();
}
最佳答案
使用AggregateBy同时聚合多个字段:
SumBy impSum = new SumBy(new Fields("i"), new Fields("ti"), long.class);
SumBy clickSum = new SumBy(new Fields("c"), new Fields("tc"), long.class);
assembly = new AggregateBy("totals", Pipe.pipes(assembly), new Fields("o", "a", "f"), 2, impSum, clickSum);
关于java - 级联 - 合并 2 个聚合,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/7009193/