Java Spark Dataset列筛选与指定列最大值获取技术求助

没问题！我会一步步教你怎么实现需求，完全不用怕不熟悉Spark~

首先先假设你的原始Dataset输出是这样的（如果和实际不符，你只需要调整代码里的字段名就行）：

原始Dataset执行dataset.show()的输出：

id	value
1	a
1	b
2	c
3	d
3	e
3	f

你的目标输出应该是带有全局递增rownum的结果，并且需要拿到rownum的最大值（示例里是6）：

目标输出：

id	value	rownum
1	a	1
1	b	2
2	c	3
3	d	4
3	e	5
3	f	6

下面是完整的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 org.apache.spark.sql.functions;

public class SparkRownumHandler {
    public static void main(String[] args) {
        // 1. 初始化SparkSession（Spark应用的核心入口）
        SparkSession spark = SparkSession.builder()
                .appName("AddRownumAndGetMax")
                .master("local[*]") // 本地调试用这个，上线部署时删掉这行
                .getOrCreate();

        // 2. 这里是模拟你的原始Dataset，实际中你可以替换成自己的数据源（比如CSV、数据库读取）
        Dataset<Row> originalDs = spark.createDataFrame(
                spark.sparkContext().parallelize(java.util.Arrays.asList(
                        org.apache.spark.sql.RowFactory.create(1, "a"),
                        org.apache.spark.sql.RowFactory.create(1, "b"),
                        org.apache.spark.sql.RowFactory.create(2, "c"),
                        org.apache.spark.sql.RowFactory.create(3, "d"),
                        org.apache.spark.sql.RowFactory.create(3, "e"),
                        org.apache.spark.sql.RowFactory.create(3, "f")
                )),
                org.apache.spark.sql.types.DataTypes.createStructType(java.util.Arrays.asList(
                        org.apache.spark.sql.types.DataTypes.createStructField("id", org.apache.spark.sql.types.DataTypes.IntegerType, false),
                        org.apache.spark.sql.types.DataTypes.createStructField("value", org.apache.spark.sql.types.DataTypes.StringType, false)
                ))
        );

        // 3. 定义全局排序窗口：按你需要的规则排序来生成rownum（这里用id+value排序，你可以改成自己的字段）
        WindowSpec globalWindow = Window.orderBy("id", "value");

        // 4. 给Dataset添加rownum列：用row_number()函数生成连续递增的行号
        Dataset<Row> targetDs = originalDs.withColumn("rownum", functions.row_number().over(globalWindow));

        // 5. 查看目标输出，和你想要的结果一致
        System.out.println("目标输出：");
        targetDs.show();

        // 6. 获取rownum的最大值
        Long maxRownum = targetDs.agg(functions.max("rownum")).first().getLong(0);
        System.out.println("rownum的最大值是：" + maxRownum);

        // 7. 关闭SparkSession
        spark.stop();
    }
}

关键说明： #

• 如果你的原始Dataset有其他字段，完全不用担心，withColumn只会新增rownum列，不会修改原有数据。
• 如果你需要的是分组内的rownum（比如每个id组内从1开始计数），只需要把窗口定义改成这样：

  WindowSpec groupWindow = Window.partitionBy("id").orderBy("value");
  

  这时候每个id组内的rownum会从1开始递增，全局最大值就是分组里最大的那个行号。
• row_number()生成的是连续的整数，所以全局rownum的最大值其实等于你的Dataset总行数，但用agg(max("rownum"))是最稳妥的获取方式。

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