如何基于sparklyr并行执行多日期区间的自定义计算函数？

我目前用sparklyr处理大型数据集，需要基于多组起止日期计算指标，每组日期对应生成一个结果数据集。最初用dplyr+mclapply实现并行，数据量增长后转用Spark，处理大数据时耗时从8小时缩短到3小时，但处理小数据集时，用for循环替代mclapply导致耗时增加。想请教是否能基于sparklyr后端并行调用自定义函数，还是只能按示例代码那样顺序执行？

示例代码：

transactions <- data.frame(date = c(as.Date('2022-01-01'),as.Date('2021-01-01'),as.Date('2020-01-01')),
                           transaction_value = c(10,20,10))

transactions_tbl <- copy_to(sc, transactions)

start_dates <- c(as.Date('2022-01-01'), as.Date('2022-01-01'), as.Date('2022-01-01'))
end_dates <- c(as.Date('2022-12-31'), as.Date('2022-12-31'), as.Date('2022-12-31'))

tot_val_func <- function(transactions_tbl,start_date, end_date) {
  year_tot <- transactions_tbl %>% 
    dplyr::filter(date >= start_date & date <= end_date) %>%
    dplyr::summarise(total_val = sum(transaction_value)) %>%
    dplyr::collect()
  
  return(year_tot)
}

for (i in 1:length(start_dates)) {
  total_val_func(transactions_tbl,start_dates[i],end_dates[i])
  # 转换后导出到数据仓库
}

方法1：利用Spark分布式能力批量处理（最优方案） #

不要用循环多次提交Spark任务，而是把所有日期范围打包成Spark表，通过一次分布式计算完成所有分组统计，这是最贴合Spark设计思路的方式，无论大小数据集都能高效处理：

# 构造带唯一标识的日期范围表
date_ranges <- data.frame(
  range_id = seq_along(start_dates),
  start_date = start_dates,
  end_date = end_dates
)
date_ranges_tbl <- copy_to(sc, date_ranges)

# 关联交易表与日期范围表，按范围分组计算
batch_results <- transactions_tbl %>%
  # 交叉关联所有日期范围
  dplyr::cross_join(date_ranges_tbl) %>%
  # 筛选符合当前范围的交易
  dplyr::filter(date >= start_date & date <= end_date) %>%
  # 按日期范围分组统计
  dplyr::group_by(range_id, start_date, end_date) %>%
  dplyr::summarise(total_val = sum(transaction_value, na.rm = TRUE)) %>%
  dplyr::collect()

# 后续可按range_id拆分结果，导出到数据仓库

这种方式只需要一次Spark作业，避免了循环中多次建立查询的开销，大数据下保持高效，小数据集也不会因为循环拖慢速度。

方法2：用R并行工具配合sparklyr（保留原函数结构） #

如果必须保留tot_val_func的结构，可以用foreach+future实现R端的并行提交Spark任务，但要注意Spark连接的复用：

library(foreach)
library(future)
library(future.apply)

# 设置并行后端（比如multisession）
plan(multisession, workers = 3)

# 并行调用函数
parallel_results <- future_map2(start_dates, end_dates, function(s, e) {
  tot_val_func(transactions_tbl, s, e)
})

# 合并结果
combined_results <- dplyr::bind_rows(parallel_results)

注意：这种方式是在R端并行提交多个Spark任务，适合日期范围组数较多但单组计算逻辑复杂的场景，但小数据集下的开销可能比方法1大，因为每个任务都要和Spark集群交互。

方法3：分场景适配（兼顾大小数据集） #

可以根据交易数据的大小自动切换处理逻辑：

• 当数据量小时，将transactions_tbl collect到本地，用mclapply并行计算
• 当数据量大时，用方法1的Spark批量处理

# 获取Spark表的行数
row_count <- transactions_tbl %>% dplyr::count() %>% dplyr::pull()

if (row_count < 100000) { # 自定义阈值
  # 小数据集：本地并行
  local_transactions <- dplyr::collect(transactions_tbl)
  local_results <- mclapply(seq_along(start_dates), function(i) {
    local_transactions %>%
      dplyr::filter(date >= start_dates[i] & date <= end_dates[i]) %>%
      dplyr::summarise(total_val = sum(transaction_value))
  })
} else {
  # 大数据集：Spark批量处理
  date_ranges <- data.frame(
    range_id = seq_along(start_dates),
    start_date = start_dates,
    end_date = end_dates
  )
  date_ranges_tbl <- copy_to(sc, date_ranges)
  
  local_results <- transactions_tbl %>%
    dplyr::cross_join(date_ranges_tbl) %>%
    dplyr::filter(date >= start_date & date <= end_date) %>%
    dplyr::group_by(range_id) %>%
    dplyr::summarise(total_val = sum(transaction_value)) %>%
    dplyr::collect()
}

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