使用 R 目标将新数据附加到现有数据

Question

我正在使用targets 工作流管道。此管道的一部分是监视 csv 文件目录的更新。这个目录有10000多个csv文件，每周都会增加新文件。我希望能够识别新添加的文件并将它们附加到现有的一组 *.rds 文件中。最简单的方法是每周重新运行创建*.rds 文件的5 个子集的过程，但这需要时间。有效的方法是识别新添加的文件，只需将bind_rows 与正确的rds 文件一起使用。

我可以通过使用dir() 和setdiff() 的典型编程很容易地做到这一点，我在其中存储了前一天的 csv 文件路径的快照。但我正在努力在targets 框架内实现这一目标。

这是一个似乎行不通的尝试。我想我想监视/_targets 目录中的临时结果，但我不知道该怎么做。而且，targets 文档建议不要在目标配置本身中使用tar_load。

tar_script({
   list(
      tar_target(csv_directory, "/csv/"),
      tar_target(csv_snapshot, dir(csv_directory)),
      tar_target(append_action, if(length(setdiff(dir(csv_directory), dir(csv_snapshot))) > 0){
                                ...}
})

Answer 1

一些可能有帮助的组件：

1. 文件目标：https://books.ropensci.org/targets/files.html。使用tar_target(format = "file")，包会监视输入和/或输出文件的更改并重新运行受影响的目标（如果有的话）。
2. 替代存储格式：https://docs.ropensci.org/targets/reference/tar_target.html#storage-formats。与其将 CSV 文件聚合到外部 RDS 文件中，不如使用tar_target(format = "feather") 之类的东西更有效，因此targets 会自动压缩您的输出数据并确保您不必担心文件的微观管理。
3. 动态分支：books.ropensci.org/targets/dynamic.html。动态分支是一种在管道运行时定义大量新目标的方法。例如，这使您可以为一个文件或一批现有文件创建一个新目标。
4. 批处理：https://books.ropensci.org/targets/dynamic.html#batching。 10000 个目标太多了，targets 包可能会因为数量太多而减慢速度，因为每个目标都会产生开销。

因此，我建议您将 CSV 文件组织成批次（例如，按周）并动态地分批处理它们。根据您的用例的具体情况，另一种批处理结构可能更合适。

csv/
├── week1/
│   ├── data1.csv
│   ├── data2.csv
│   ├── ...
├── week2/
│   ├── data1.csv
│   ├── data2.csv
│   ├── ...
...

管道示意图：

# _targets.R
process_csv_dir <- function(csv_dir) {...} # custom user-defined function
list(
  tar_target(csv_dir, list.files("csv", full.names = TRUE)),
  tar_target(
    processed_data,
    process_csv_dir(csv_dir),
    pattern = map(csv_dir), # dynamic branching
    format = "feather" # from the arrow package
  )
)