R Sparklyr中base::cumsum()与dplyr::mutate_at()联用失效问题

问题背景 #

需对已分组排序的Spark DataFrame多列计算累积求和，使用mutate_at(.vars, .funs = cumsum)时报错object 'mpg' not found，但替换为sum()、单独使用mutate("mpg_cumulative" = cumsum(mpg))或mutate_at(.vars = 'mpg', .funs = ~cumsum(.))均可正常运行。环境配置：

• R version 4.4.1 (2024-06-14)
• Spark 3.5.0
• sparklyr_1.8.4、dbplyr_2.4.0、dplyr_1.1.4

报错原因 #

直接传递base::cumsum给mutate_at的.funs参数时，dbplyr无法将其正确绑定到Spark DataFrame的列上下文，导致无法解析列名。而lambda表达式~cumsum(.)会明确将函数与当前处理的列关联，让dbplyr能正确生成对应的窗口SQL执行逻辑。

解决方案 #

方案1：使用across()替代mutate_at()（推荐） #

dplyr 1.0+版本已推荐用across()替代旧的mutate_at/mutate_all系列函数，语法更清晰且兼容性更强，支持直接传入多列向量批量处理：

sdf_mtcars %>% 
  group_by(cyl) %>%
  window_order(disp) %>%
  mutate(across(c('mpg', 'hp'), cumsum)) %>% # 可添加更多列到向量中
  ungroup() %>% 
  collect()

若需保留原列并生成带后缀的累积和列，可通过.names参数重命名：

sdf_mtcars %>% 
  group_by(cyl) %>%
  window_order(disp) %>%
  mutate(across(c('mpg', 'hp'), ~cumsum(.), .names = "{col}_cum")) %>%
  ungroup() %>% 
  collect()

方案2：优化mutate_at()的函数传递方式 #

如果坚持使用mutate_at，可将函数包装为列表传递，让dbplyr正确解析：

sdf_mtcars %>% 
  group_by(cyl) %>%
  window_order(disp) %>%
  mutate_at(.vars = c('mpg'), .funs = list(cumsum)) %>%
  ungroup() %>% 
  collect()

也可通过命名列表指定新列名：

sdf_mtcars %>% 
  group_by(cyl) %>%
  window_order(disp) %>%
  mutate_at(.vars = c('mpg'), .funs = list(mpg_cum = cumsum)) %>%
  ungroup() %>% 
  collect()

方案3：显式调用dbplyr窗口函数 #

直接使用dbplyr::cumsum显式指定窗口函数，避免base函数的解析歧义：

sdf_mtcars %>% 
  group_by(cyl) %>%
  window_order(disp) %>%
  mutate(across(c('mpg'), dbplyr::cumsum)) %>%
  ungroup() %>% 
  collect()

验证结果 #

以上方案均能在指定环境下正常运行，正确生成分组排序后的累积和结果。其中across()是当前dplyr的标准写法，扩展性和可读性最优。

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