求助：用R循环从AWS S3批量加载文件并写入SQLite表

问题背景 #

之前手动处理8个S3文件时，先将第一个文件覆盖写入SQLite表，剩余文件逐个追加。现在文件数量达到1154个，需要用循环批量实现相同逻辑，但自行编写的循环无法正常运行。

修正后的完整代码 #

require(aws.s3)
require(data.table)
require(tidyverse)
require(rjson)
require(parallel)
require(DBI)

# 设置SQLite连接 -----------------------------------------------------------
datamart <- dbConnect(RSQLite::SQLite(), "datamart.sqlite")

# 配置AWS S3凭证 -----------------------------------------------------------
Sys.setenv(
  "AWS_ACCESS_KEY_ID" = 'HIDDEN',
  "AWS_SECRET_ACCESS_KEY" = 'HIDDEN',
  "AWS_DEFAULT_REGION" = "us-east-1")

# 获取S3存储桶中的目标文件列表 ------------------------------------------------
bucket_name <- "thg_bucket"
prefix_path <- "folder1/folder2/"

# 提取完整的文件Key，过滤掉前缀目录本身（避免空条目）
file_keys <- get_bucket(bucket_name, prefix = prefix_path) %>%
  data.table::rbindlist() %>%
  filter(Key != prefix_path) %>%  # 排除前缀对应的目录条目
  pull(Key)  # 提取为字符向量，而非列表

# 循环读取文件并写入SQLite ------------------------------------------------
for(i in seq_along(file_keys)){
  temp_file <- tempfile()
  # 拼接完整的S3对象路径
  s3_object_path <- paste0("s3://", bucket_name, "/", file_keys[i])
  
  # 从S3下载文件到临时文件
  save_object(object = s3_object_path, file = temp_file)
  
  # 读取并处理数据
  pol_temp <- read.csv(file = temp_file, sep = "|", quote = NULL) %>%
    mutate(across(everything(), ~ map_chr(.x, ~ str_sub(string = .x, start = 2, end = -2))))
  
  # 写入SQLite：第一个文件覆盖，后续追加
  if(i == 1){
    dbWriteTable(datamart, "policy", pol_temp, overwrite = TRUE)
  } else {
    dbWriteTable(datamart, "policy", pol_temp, append = TRUE)
  }
  
  # 清理临时变量
  rm(pol_temp)
  file.remove(temp_file)  # 主动删除临时文件，释放空间
}

# 关闭数据库连接
dbDisconnect(datamart)

关键修正点 #

• 文件列表处理优化：原代码将文件Key转为列表导致路径拼接错误，改为提取为字符向量；同时过滤掉前缀对应的目录条目，避免处理空文件
• 写入逻辑修正：新增判断分支，第一个文件使用overwrite = TRUE，后续文件使用append = TRUE，匹配需求
• 资源清理优化：主动删除临时文件避免磁盘占用，循环结束后关闭数据库连接释放资源
• 路径拼接可靠性提升：通过桶名+Key的方式拼接S3路径，避免手动前缀拼接的字符错误

可选优化建议 #

• 并行处理：利用parallel包实现并行下载和处理，提升大数量文件的处理速度
• 分批写入：批量读取多个文件合并后再写入，减少数据库IO次数
• 错误捕获：添加tryCatch块捕获单个文件处理失败的情况，避免整个循环中断

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