Group by inside clause子句，spark java

Question

我在SparkJava (IntelliJ应用程序)中就有这个过程，在那里我遇到了一个我还不知道如何解决的问题。首先，我声明数据集：

private static final String CONTRA1 = "contra1";

query = "select contra1, ..., eadfinal, , ..., data_date" + FROM + dbSchema + TBLNAME " + WHERE fech = '" + fechjmCto2 + "' AND s1emp=49";
        Dataset<Row> jmCto2 = sql.sql(query);

然后我进行了计算，我分析了一些字段来分配一些字面值。我的问题出在aggegate函数中：

Dataset<Row> contrCapOk1 = contrCapOk.join(jmCto2,
        contrCapOk.col(CONTRA1).equalTo(jmCto2.col(CONTRA1)),LEFT)
        .select(contrCapOk.col("*"),
        jmCto2.col("ind"),
 
functions.when(jmCto2.col(CONTRA1).isNull(),functions.lit(NUEVES))
     .when(jmCto2.col("ind").equalTo("N"),functions.lit(UNOS))
     .otherwise(jmCto2.groupBy(CONTRA1).agg(functions.sum(jmCto2.col("eadfinal")))).as("EAD"),

我想要的是在其他部分中求和。但是当我执行集群时，在日志中给我这条消息。

User class threw exception: java.lang.RuntimeException: Unsupported literal type class org.apache.spark.sql.Dataset [contra1: int, sum(eadfinal): decimal(33,6)] 

在第211行中，否则行。

你知道问题出在哪里吗？

谢谢。

Answer 1

不能在column子句中使用groupBy和聚合函数。要做你想做的事情，你必须使用window。

对于您的情况，您可以定义以下窗口：

import org.apache.spark.sql.expressions.Window;
import org.apache.spark.sql.expressions.WindowSpec;

...

WindowSpec window = Window
  .partitionBy(CONTRA1)
  .rangeBetween(Window.unboundedPreceding(), Window.unboundedFollowing());

哪里

• partitionBy等同于aggregation
• rangeBetween的groupBy确定聚合函数将使用分区的哪些行，这里我们使用

的所有行

然后在调用聚合函数时使用此窗口，如下所示：

import org.apache.spark.sql.functions;

...

Dataset<Row> contrCapOk1 = contrCapOk.join(
    jmCto2,
    contrCapOk.col(CONTRA1).equalTo(jmCto2.col(CONTRA1)),
    LEFT
  )
  .select(
    contrCapOk.col("*"),
    jmCto2.col("ind"),
    functions.when(jmCto2.col(CONTRA1).isNull(), functions.lit(NUEVES))
     .when(jmCto2.col("ind").equalTo("N"), functions.lit(UNOS))
     .otherwise(functions.sum(jmCto2.col("eadfinal")).over(window))
     .as("EAD")
  )