如何在R targets框架中识别新增csv并将数据追加到现有rds文件

• 利用targets的cue机制强制每次运行时扫描CSV目录，无需手动维护快照对比逻辑
• 新增独立持久化文件存储已处理CSV路径列表，避免直接访问/_targets内部临时文件
• 所有依赖通过target入参传递，无需在配置内调用tar_load，符合官方规范

完整实现代码 #

library(targets)
library(tarchetypes)
library(tidyverse)

# 预定义持久化存储路径：已处理CSV列表、分组RDS文件存储目录
processed_csv_path <- "processed_csv_records.rds"
rds_output_dir <- "./rds_subsets/"

# 首次运行前初始化（仅需执行一次）
if (!dir.exists(rds_output_dir)) dir.create(rds_output_dir)
# 若5组RDS文件尚未生成，首次运行前先写入空表模板
# for (i in 1:5) saveRDS(tibble(), paste0(rds_output_dir, "group_", i, ".rds"))
if (!file.exists(processed_csv_path)) saveRDS(character(0), processed_csv_path)

tar_script({
  list(
    # 1. 每次运行强制扫描CSV目录，获取全量CSV路径
    tar_target(
      name = all_csv_files,
      command = list.files(path = "/csv/", pattern = "\\.csv$", full.names = TRUE),
      cue = tar_cue(mode = "always")
    ),
    # 2. 读取已处理CSV路径列表
    tar_target(
      name = processed_csvs,
      command = readRDS(processed_csv_path),
      cue = tar_cue(mode = "always")
    ),
    # 3. 计算新增CSV路径集合
    tar_target(
      name = new_csv_files,
      command = setdiff(all_csv_files, processed_csvs)
    ),
    # 4. 动态分支批量读取新增CSV数据
    tar_target(
      name = new_raw_data,
      command = read_csv(new_csv_files, col_types = cols(.default = col_guess())),
      pattern = map(new_csv_files),
      iteration = "list"
    ),
    # 5. 合并新增数据、追加到对应RDS、更新已处理列表
    tar_target(
      name = append_update_task,
      command = {
        # 无新增文件时直接跳过
        if (length(new_raw_data) == 0) return(invisible(TRUE))
        
        # 合并所有新增数据
        all_new_df <- bind_rows(new_raw_data)
        
        # 替换为你自己的5组RDS拆分追加逻辑，示例按group字段拆分
        group_split_df <- group_split(all_new_df, group_field, .keep = TRUE)
        walk(group_split_df, function(sub_df) {
          group_id <- unique(sub_df$group_field)
          target_rds <- paste0(rds_output_dir, "group_", group_id, ".rds")
          old_df <- readRDS(target_rds)
          updated_df <- bind_rows(old_df, sub_df)
          saveRDS(updated_df, target_rds)
        })
        
        # 更新已处理CSV列表
        updated_processed <- c(processed_csvs, new_csv_files)
        saveRDS(updated_processed, processed_csv_path)
        
        return(TRUE)
      }
    )
  )
})

注意事项 #

• 首次运行前需要完成初始化步骤，生成空的5组RDS模板和已处理记录文件，首次运行会全量处理所有存量CSV，后续运行仅处理新增文件
• 可根据实际需求调整CSV读取的列类型、RDS分组逻辑，动态分支支持并行处理大幅提升新增文件读取效率

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