新版Apache Spark（Java）实现等价SQL分组聚合查询方法

我来帮你梳理下怎么用新版Apache Spark（3.x及以上版本）的Java API实现和这条SQL等价的分组聚合操作。先把需求对应清楚：你的SQL是按id和交易日期的月份分组，计算每个分组的交易金额总和；而数据集里的字段ID对应SQL的id，Spend对应thistransaction，DateTime对应transactiondisplaydate。

接下来分两种常用的实现方式：

方式一：使用DataFrame API（推荐，灵活且贴近SQL逻辑） #

这种方式不需要定义实体类，直接基于DataFrame做字段转换、分组聚合，和SQL逻辑的对应关系非常直观：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

public class SparkGroupByAgg {
    public static void main(String[] args) {
        // 初始化SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("GroupByAggExample")
                .master("local[*]") // 本地模式，生产环境请移除
                .getOrCreate();

        // 读取示例数据（假设是CSV格式，根据实际存储调整）
        Dataset<Row> df = spark.read()
                .option("header", "false") // 如果没有表头的话
                .option("inferSchema", "false") // 手动指定类型更可靠
                .csv("path/to/your/history/data");

        // 重命名字段并转换类型：对应SQL里的类型转换
        Dataset<Row> transformedDF = df
                .withColumnRenamed("_c0", "id")
                .withColumnRenamed("_c1", "spend")
                .withColumnRenamed("_c2", "transactiondisplaydate")
                // 将Spend转为Decimal类型，对应SQL的thistransaction::decimal
                .withColumn("spend", col("spend").cast("decimal(10,2)"))
                // 将日期字符串转为Timestamp，对应SQL的transactiondisplaydate::timestamp
                .withColumn("transactiondisplaydate", col("transactiondisplaydate").cast("timestamp"))
                // 提取月份，对应SQL的date_part('month', ...) as month
                .withColumn("month", date_part(lit("month"), col("transactiondisplaydate")));

        // 执行分组聚合，和SQL的group by + sum完全对应
        Dataset<Row> resultDF = transformedDF
                .groupBy("id", "month")
                .agg(sum("spend").alias("total_spend"));

        // 展示结果
        resultDF.show();

        // 停止SparkSession
        spark.stop();
    }
}

代码说明： #

• date_part(lit("month"), col("transactiondisplaydate")) 完全等价于SQL里的date_part('month', transactiondisplaydate::timestamp)，会返回月份的数字（比如9代表9月）。
• sum("spend").alias("total_spend") 对应SQL的sum(thistransaction::decimal)，并给结果列起别名。
• 字段重命名是为了让代码更贴合SQL里的字段名，如果你读取的数据源已经有对应表头，可以跳过重命名步骤。

方式二：使用强类型Dataset（适合有明确数据结构的场景） #

如果你的业务逻辑需要强类型校验，可以定义对应的POJO类，用Dataset来操作：

1. 定义实体类 #

import java.math.BigDecimal;
import java.sql.Timestamp;

public class Transaction {
    private String id;
    private BigDecimal spend;
    private Timestamp transactiondisplaydate;

    // 必须要有无参构造函数
    public Transaction() {}

    // 全参构造函数、getter和setter
    public Transaction(String id, BigDecimal spend, Timestamp transactiondisplaydate) {
        this.id = id;
        this.spend = spend;
        this.transactiondisplaydate = transactiondisplaydate;
    }

    public String getId() { return id; }
    public void setId(String id) { this.id = id; }

    public BigDecimal getSpend() { return spend; }
    public void setSpend(BigDecimal spend) { this.spend = spend; }

    public Timestamp getTransactiondisplaydate() { return transactiondisplaydate; }
    public void setTransactiondisplaydate(Timestamp transactiondisplaydate) { this.transactiondisplaydate = transactiondisplaydate; }
}

2. 实现分组聚合的代码 #

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

public class StrongTypeGroupByAgg {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("StrongTypeGroupByAgg")
                .master("local[*]")
                .getOrCreate();

        // 读取数据并转换为强类型Dataset
        Dataset<Transaction> transactionDS = spark.read()
                .option("header", "false")
                .schema(org.apache.spark.sql.Encoders.bean(Transaction.class).schema())
                .csv("path/to/your/history/data")
                .as(org.apache.spark.sql.Encoders.bean(Transaction.class));

        // 提取月份并聚合
        Dataset<Row> resultDF = transactionDS
                .withColumn("month", date_part(lit("month"), col("transactiondisplaydate")))
                .groupBy("id", "month")
                .agg(sum("spend").alias("total_spend"));

        resultDF.show();

        spark.stop();
    }
}

验证结果 #

用你提供的示例数据运行后，结果会类似这样：

+-------+-----+-----------+
|     id|month|total_spend|
+-------+-----+-----------+
|468429 |9   |53.30      |
|520003 |9   |239.70     |
+-------+-----+-----------+

内容的提问来源于stack exchange，提问作者user3099077