Spark Java基于自定义条件的分组过滤优化方案问询

AI 小北

2026-7-27

更简洁的实现方式

可以通过两种更简洁的方案实现需求，完全替代原代码中手动处理数组的复杂逻辑，用Spark内置函数或SQL风格表达式完成：

方法一：子查询筛选用户后关联原表

先通过聚合找出存在分数>20的用户，再关联原数据集保留这些用户的所有记录：

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

// 读取原始数据并转换Score为数值类型
Dataset<Row> rawDf = sparkSession.read()
    .csv("scores.csv")
    .toDF("Name", "Score")
    .withColumn("Score", col("Score").cast("long"));

// 筛选出存在高分记录的用户
Dataset<Row> validUsers = rawDf.groupBy("Name")
    .agg(max("Score").as("max_score"))
    .filter(col("max_score") > 20)
    .select("Name");

// 关联原表得到最终结果
rawDf.join(validUsers, "Name")
    .show(false);

方法二：使用窗口函数标记过滤

通过窗口函数给每个用户的所有记录添加标记，直接过滤出符合条件的记录：

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

Dataset<Row> rawDf = sparkSession.read()
    .csv("scores.csv")
    .toDF("Name", "Score")
    .withColumn("Score", col("Score").cast("long"));

// 定义按用户分组的窗口
WindowSpec userWindow = Window.partitionBy("Name");

// 添加标记列并过滤
rawDf.withColumn("has_high_score", max(when(col("Score") > 20, 1).otherwise(0)).over(userWindow))
    .filter(col("has_high_score") == 1)
    .drop("has_high_score")
    .show(false);