如何实现可跨目录中多个文件的Rolling window函数？

问题背景 #

我有一个包含多个文件的目录，文件按字母顺序排列，这决定了分析顺序；每个文件内的ID列指定了数据的分析顺序。需要用自定义滚动窗口函数完成跨所有文件的连续计算，仅在第一个文件开头用NA填充，窗口要能跨文件边界（即前一个文件末尾和后一个文件开头）。但文件过大，无法同时加载两个以上文件到内存，也不能读整个数据集。需要一个通用方案，适配不同函数和窗口大小，本次以窗口大小为2的均值计算为例。

生成测试数据 #

以下代码用于生成5个测试数据框并保存到指定目录：

# 创建测试数据框A（ID 1-10）
A <- data.frame(
  ID = seq(from=1, to=10, by=1),
  Value=sample(1:10, size=10, replace=TRUE))
# 创建测试数据框B（ID 11-20）
B <- data.frame(
  ID = seq(from=11, to=20, by=1),
  Value=sample(1:10, size=10, replace=TRUE))
# 创建测试数据框C（ID 21-30）
C <- data.frame(
  ID = seq(from=21, to=30, by=1),
  Value=sample(1:10, size=10, replace=TRUE))
# 创建测试数据框D（ID 31-40）
D <- data.frame(
  ID = seq(from=31, to=40, by=1),
  Value=sample(1:10, size=10, replace=TRUE))
# 创建测试数据框E（ID 41-50）
E <- data.frame(
  ID = seq(from=41, to=50, by=1),
  Value=sample(1:10, size=10, replace=TRUE))

# 创建目标目录
dir.create("/home/phil/Desktop/test")

# 将数据框写入CSV文件
write_csv(A, "/home/phil/Desktop/test/A.csv", col_names = TRUE)
write_csv(B, "/home/phil/Desktop/test/B.csv", col_names = TRUE)
write_csv(C, "/home/phil/Desktop/test/C.csv", col_names = TRUE)
write_csv(D, "/home/phil/Desktop/test/D.csv", col_names = TRUE)
write_csv(E, "/home/phil/Desktop/test/E.csv", col_names = TRUE)

# 清理内存中的数据框
rm(A, B, C, D, E)

# 获取目录下所有CSV文件的完整路径（按字母顺序）
dfList <- list.files("/home/phil/Desktop/test", pattern = "*.csv", full.names=TRUE)

核心思路 #

由于内存限制，每次仅加载当前文件和前一个文件的末尾片段（长度为窗口大小-1），实现跨文件窗口计算：

• 按顺序遍历文件，每次处理当前文件时，结合前一个文件的末尾数据
• 对合并后的临时数据应用滚动窗口函数
• 提取当前文件对应的计算结果（剔除前文件末尾的冗余结果）
• 保留当前文件的末尾数据，用于下一个文件的边界计算
• 第一个文件单独处理，开头按窗口大小填充NA

通用滚动窗口计算实现 #

library(readr)
library(dplyr)
library(slider) # 可替换为其他滚动计算工具

# 定义通用跨文件滚动计算函数
cross_file_rolling <- function(file_list, window_size, rolling_func, output_dir = NULL) {
  prev_tail <- NULL # 存储前一个文件的末尾数据

  for (i in seq_along(file_list)) {
    current_file <- file_list[i]
    current_df <- read_csv(current_file, show_col_types = FALSE)

    if (i == 1) {
      # 处理第一个文件，开头填充NA
      current_df <- current_df %>%
        mutate(Rolling_Result = rolling_func(Value, window = window_size, na.rm = FALSE))
      
      # 保存当前文件的末尾数据（窗口大小-1行）
      prev_tail <- tail(current_df, window_size - 1)
    } else {
      # 合并前文件末尾数据与当前文件，实现跨边界窗口
      combined_df <- bind_rows(prev_tail, current_df)
      combined_df <- combined_df %>%
        mutate(Rolling_Result = rolling_func(Value, window = window_size, na.rm = FALSE))
      
      # 提取当前文件对应的结果（移除前文件末尾的行）
      current_df <- slice(combined_df, -(1:(window_size - 1)))
      # 更新前文件末尾数据为当前文件的末尾片段
      prev_tail <- tail(current_df, window_size - 1)
    }

    # 输出结果到指定目录（可选）
    if (!is.null(output_dir)) {
      write_csv(current_df, file.path(output_dir, basename(current_file)), col_names = TRUE)
    }

    # 清理内存
    rm(current_df)
    gc()
  }
  invisible(TRUE)
}

# 示例：窗口大小为2的滚动均值函数
rolling_mean_2 <- function(x, window, na.rm) {
  slide_dbl(x, mean, .before = window - 1, .complete = TRUE)
}

# 创建输出目录
output_dir <- "/home/phil/Desktop/test_output"
dir.create(output_dir, showWarnings = FALSE)

# 执行跨文件滚动计算
cross_file_rolling(dfList, window_size = 2, rolling_func = rolling_mean_2, output_dir = output_dir)

方案说明 #

• 通用性：只需替换rolling_func参数即可实现不同滚动计算（如滚动求和、最大值等），只需保证函数接收x（待计算向量）、window（窗口大小）、na.rm（是否忽略NA）三个参数，返回对应结果向量。
• 内存控制：每次仅加载当前完整文件和前一个文件的window_size-1行数据，内存占用极低。
• 跨文件连续性：通过合并前文件末尾与当前文件，确保窗口能跨边界计算，结果无断裂。

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