时间戳

首页 > 解决方案 > Spark java：在多个列上聚合并重命名它们

java - Spark java：在多个列上聚合并重命名它们

问题描述

我想在我之前不知道的多个列上对我的数据集进行分组，因此.agg()允许传递一个Map，其中键是列名，值是聚合名称，例如例如我可以这样做：

    for(String column:columns)
        map.put(column, "sum");
    ds.groupBy("someColumn").agg(map)

到这里为止都很好，但我想保留原来的列名并且没有这样的东西

'|sum(column1)|sum(column12)|...'

我试过这样做，但没有奏效：

map.put(column, "sum alias " + column);

可以用java api做到这一点吗？

标签： javaapache-sparkapache-spark-sqlaggregate-functionsdynamic-columns

解决方案

试试这个-

我已将列名作为别名提供给sum(column)

    Dataset<Row> df = spark.range(2).withColumn("value", lit(2));
        df.show(false);
        df.printSchema();

        /**
         * +---+-----+
         * |id |value|
         * +---+-----+
         * |0  |2    |
         * |1  |2    |
         * +---+-----+
         *
         * root
         *  |-- id: long (nullable = false)
         *  |-- value: integer (nullable = false)
         */
        Map<String, String> map = new HashMap<>();
        for(String column:df.columns())
            map.put(column, "sum");

        List<Column> cols = map.entrySet().stream().map(c -> expr(String.format("%s(%s) as %s", c.getValue(), c.getKey(), c.getKey())))
                .collect(Collectors.toList());


        df.agg(cols.get(0), toScalaSeq(cols.subList(1, cols.size()))).show(false);
        /**
         * +---+-----+
         * |id |value|
         * +---+-----+
         * |1  |4    |
         * +---+-----+
         */

效用-

 <T> Buffer<T> toScalaSeq(List<T> list) {
        return JavaConversions.asScalaBuffer(list);
    }

推荐阅读