基于Java Spark DataFrame按id去重并保留最新版本的最优方案

在Spark里处理这类「保留重复ID的最新版本数据」的需求，其实有几种高效的实现方式，完全不用复杂的嵌套SQL，我来给你详细拆解下：

一、Spark DataFrame API（Java）实现方式（最优推荐） #

首先最推荐的是用Spark的DataFrame API结合窗口函数来实现，这也是业界处理这类「分组取TopN」场景的标准最优方案，代码不仅类型安全，逻辑也一目了然，性能还拉满。

实现逻辑 #

1. 定义窗口规则：按id字段分区，把同一个ID的所有数据归为一组；再按版本字段（比如version）降序排序，这样每个ID组里最新版本的数据会排在第一位。
2. 给每条数据添加行号：同一ID组内，最新版本的行号会是1，旧版本依次递增。
3. 过滤出行号为1的数据，就是我们要的仅保留最新版本的结果。

Java代码示例 #

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

public class KeepLatestVersionDemo {
    public static void main(String[] args) {
        // 初始化SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("KeepLatestVersion")
                .master("local[*]") // 本地调试用，生产环境可去掉
                .getOrCreate();

        // 读取原始数据，这里假设是JSON格式，你可以换成自己的数据源（比如Parquet、JDBC等）
        Dataset<Row> originalDF = spark.read().json("path/to/your/data");

        // 定义窗口：按id分区，version降序排序
        WindowSpec windowSpec = Window.partitionBy("id").orderBy(desc("version"));

        // 添加行号、过滤最新版本、移除临时行号字段
        Dataset<Row> latestVersionDF = originalDF
                .withColumn("row_num", row_number().over(windowSpec))
                .filter(col("row_num").equalTo(1))
                .drop("row_num");

        // 查看结果
        latestVersionDF.show();
        spark.stop();
    }
}

这个方案的优势 #

• 性能高效：Spark优化器会对窗口操作做针对性优化，不需要多次Shuffle（对比传统的分组关联方式）。
• 逻辑直观：代码逻辑清晰，一眼就能看懂是按ID分组取最新版本。
• 扩展性强：如果以后需要保留前N个版本，只需要修改过滤条件为row_num <= N即可。

二、SQL方式实现（无需复杂嵌套） #

如果你更习惯用SQL来处理数据，也完全没问题，用窗口函数的SQL写法就可以，完全不需要复杂的嵌套查询，比传统的「分组取最大版本再关联」的方式更简洁高效。

简化版SQL示例（CTE写法，可读性更强） #

WITH ranked_data AS (
    SELECT 
        *,
        -- 按id分区，version降序排，给每条数据打行号
        ROW_NUMBER() OVER (PARTITION BY id ORDER BY version DESC) AS row_num
    FROM original_table
)
-- 只保留行号为1的最新版本数据，列出你需要的字段
SELECT id, version, col1, col2
FROM ranked_data
WHERE row_num = 1;

也可以写成子查询形式（不需要CTE） #

SELECT id, version, col1, col2
FROM (
    SELECT 
        *,
        ROW_NUMBER() OVER (PARTITION BY id ORDER BY version DESC) AS row_num
    FROM original_table
) t
WHERE t.row_num = 1;

这里的逻辑和API方式完全一致，只是用SQL表达而已。对比那种先分组取每个ID的最大版本，再关联原表的嵌套写法，窗口函数的SQL只需要一次扫描数据，性能更优，代码也更简洁。

关于是否需要嵌套SQL？ #

答案是不需要复杂嵌套。上面的SQL示例虽然用了子查询/CTE，但这是窗口函数的标准写法，不算复杂的嵌套逻辑，而且比传统的多层嵌套写法高效得多。

总结 #

• 最优方案是使用Spark DataFrame API（Java）结合窗口函数，类型安全、性能好、易维护。
• 如果偏好SQL，用窗口函数的SQL即可，无需复杂嵌套，简洁高效。

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