SparkR是否有等价R的unique和orderBy且保留全部列的函数？

核心逻辑对齐说明 #

原本地R代码的核心逻辑为：按CUSTOMER_NUMBER分组，取每组内BOOK_BALANCE_LCY最小、最大的第一条记录对应的CALENDAR_DATE，最终输出仅包含客户号、最小余额日期、最大余额日期的结果。

现有代码问题根因 #

• 聚合列未显式命名：agg(min(minmaxAcctDates$BOOK_BALANCE_LCY))生成的列无自定义别名，关联时无法匹配
• 未指定关联键：join操作未传入on参数定义关联字段，导致两边同名字段重复保留，触发CUSTOMER_NUMBER歧义报错
• 逻辑顺序错误：最小余额处理流程中提前调用未定义的AccountsFilteredMax对象，且冗余collect操作会将全量数据拉取到Driver节点，大数据量下直接触发OOM
• 未做类型对齐：原本地代码将CUSTOMER_NUMBER转为整数类型，Spark读取阶段未做对应转换

最优实现方案（窗口函数版，无关联性能损耗） #

直接用窗口函数分组排序取top1，避免join带来的重复列、性能问题：

library(SparkR)

# 1. 读取数据并对齐类型
minmaxAcctDates <- read.df(
  "abfss://x@adlsstorage.dfs.core.windows.net/x/Accounts_BalanceEOD.csv", 
  source = "csv", 
  header = "true", 
  inferSchema = "true"
)
# 对齐原逻辑转CUSTOMER_NUMBER为整数
minmaxAcctDates$CUSTOMER_NUMBER <- cast(minmaxAcctDates$CUSTOMER_NUMBER, "integer")

# 2. 定义窗口：按客户号分区，余额升序排序取最小余额记录
w_min <- windowPartitionBy("CUSTOMER_NUMBER") %>% orderBy("BOOK_BALANCE_LCY")
# 余额降序排序取最大余额记录
w_max <- windowPartitionBy("CUSTOMER_NUMBER") %>% orderBy(desc("BOOK_BALANCE_LCY"))

# 3. 打行号、取每组第一条记录
min_df <- minmaxAcctDates %>% 
  withColumn("rn", row_number() %>% over(w_min)) %>% 
  filter(column("rn") == 1) %>% 
  withColumnRenamed("CALENDAR_DATE", "MIN_AC_BAL_DATE") %>% 
  select("CUSTOMER_NUMBER", "MIN_AC_BAL_DATE")

max_df <- minmaxAcctDates %>% 
  withColumn("rn", row_number() %>% over(w_max)) %>% 
  filter(column("rn") == 1) %>% 
  withColumnRenamed("CALENDAR_DATE", "MAX_AC_BAL_DATE") %>% 
  select("CUSTOMER_NUMBER", "MAX_AC_BAL_DATE")

# 4. 关联两个结果得到最终表（如果需要合并为一张表的话，不需要可直接输出min_df和max_df）
final_df <- join(min_df, max_df, on = "CUSTOMER_NUMBER", how = "inner")

原写法修复版（兼容原有agg+join逻辑） #

如果要沿用先聚合再关联的原有写法，修正后代码如下：

library(SparkR)

# 读取数据+类型转换和上面一致
minmaxAcctDates <- read.df(
  "abfss://x@adlsstorage.dfs.core.windows.net/x/Accounts_BalanceEOD.csv", 
  source = "csv", 
  header = "true", 
  inferSchema = "true"
)
minmaxAcctDates$CUSTOMER_NUMBER <- cast(minmaxAcctDates$CUSTOMER_NUMBER, "integer")

# 处理最小余额日期
min_agg <- groupBy(minmaxAcctDates, "CUSTOMER_NUMBER") %>% 
  agg(min_BAL = min(minmaxAcctDates$BOOK_BALANCE_LCY))
# 显式指定关联键，避免重复列
AccountsFilteredMin <- join(min_agg, minmaxAcctDates, 
                           on = c("CUSTOMER_NUMBER", "BOOK_BALANCE_LCY" = "min_BAL"),
                           how = "inner") %>% 
  dropDuplicates("CUSTOMER_NUMBER") %>% 
  withColumnRenamed("CALENDAR_DATE", "MIN_AC_BAL_DATE") %>% 
  select("CUSTOMER_NUMBER", "MIN_AC_BAL_DATE")

# 处理最大余额日期
max_agg <- groupBy(minmaxAcctDates, "CUSTOMER_NUMBER") %>% 
  agg(max_BAL = max(minmaxAcctDates$BOOK_BALANCE_LCY))
AccountsFilteredMax <- join(max_agg, minmaxAcctDates, 
                           on = c("CUSTOMER_NUMBER", "BOOK_BALANCE_LCY" = "max_BAL"),
                           how = "inner") %>% 
  dropDuplicates("CUSTOMER_NUMBER") %>% 
  withColumnRenamed("CALENDAR_DATE", "MAX_AC_BAL_DATE") %>% 
  select("CUSTOMER_NUMBER", "MAX_AC_BAL_DATE")

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