如何在循环中生成 Spark Dataset 聚合 long expers？

Question

我正在使用 Java Spark 统计数据集。

我需要聚合一个由许多专家组成的数据集，所以代码又长又难看。 expers有一些共同的逻辑，我可以用循环生成expers吗？ 这是代码示例，实际代码有数百行重复代码：

Dataset<Row> res = ds.groupBy(ds.col("uid")).agg(functions.max(ds.col("create_time")).as("create_time"),functions.sum(functions.when(ds.col("date_diff").$less$eq(30).and(ds.col("call_type").isin(callTypeOut)),
                    ds.col("duration"))).as("caller_call_time_1"),
                          functions.sum(functions.when(ds.col("date_diff").$less$eq(60)
                            .and(ds.col("call_type").isin(callTypeOut)),
                    ds.col("duration"))).as("caller_call_time_2"),
            functions.sum(functions.when(ds.col("date_diff").$less$eq(90)
                            .and(ds.col("call_type").isin(callTypeOut)),
                    ds.col("duration"))).as("caller_call_time_3"),
            functions.sum(functions.when(ds.col("date_diff").$less$eq(120)
                            .and(ds.col("call_type").isin(callTypeOut)),
                    ds.col("duration"))).as("caller_call_time_4"),
            functions.sum(functions.when(ds.col("date_diff").$less$eq(150)
                            .and(ds.col("call_type").isin(callTypeOut)),
                    ds.col("duration"))).as("caller_call_time_5"),
            functions.sum(functions.when(ds.col("date_diff").$less$eq(180)
                            .and(ds.col("call_type").isin(callTypeOut)),
                    ds.col("duration"))).as("caller_call_time_6"))

我得到这样的解决方案：

List<Column> exprs = new ArrayList<>();
for (int i = 1; i < 7; i ++ ) {
    exprs.add(functions.sum(functions.when(ds.col("date_diff").$less$eq(30*i)
                        .and(ds.col("call_type").isin(callTypeOut)),
                ds.col("duration"))).as("caller_call_time_"+Integer.toString(i));

}
Dataset<Row> res = ds.groupBy(ds.col("uid")).agg(functions.max(ds.col("create_time")).as("create_time"),
exprs.toArray(new Column[exprs.size()]));

Answer 1

你可以：

1. 创建一个数据框（Dataset 是一个数据框，与其他任何数据集相对），之前添加所有其他列，然后对新创建的列执行聚合。您可以循环创建列。

2. 在 Java.

中构建一个 UDAF（用户定义的聚合函数）来处理您的自定义代码

希望对您有所帮助...