如何使用R通过CLI实现CSV内容的流式读取？

流式读取适用于标准输入流、管道流、超大文件等无法一次性加载全量数据到内存的场景，核心逻辑是分片读取、逐块处理，无需加载完整文件即可完成数据计算。下面给出两种常用实现方案，均兼容你原有CLI参数的使用习惯：

方案1：使用readr包分块处理（推荐，封装完善性能高） #

tidyverse生态的readr包内置了流式读取接口，支持自定义分块大小和处理逻辑，无需手动处理连接和格式解析，适配绝大多数流式场景：

library(optparse)
library(readr)

option_list <- list(
  make_option(c("-c", "--csv"), type = "character", default = "stdin",
              help = "CSV文件路径，默认读取标准输入流", metavar = "character"),
  make_option(c("-s", "--chunk-size"), type = "integer", default = 10000,
              help = "每批读取的行数，默认10000行", metavar = "integer")
)
opt_parser <- OptionParser(option_list = option_list)
opt <- parse_args(opt_parser)

# 自定义每块数据的处理逻辑，可按需修改为统计、过滤、输出等操作
process_chunk <- function(chunk, pos) {
  print(paste0("正在处理第", pos, "批数据，当前块行数：", nrow(chunk)))
  # 示例：计算目标列的均值并输出
  # cat("当前块target列均值：", mean(chunk$target, na.rm = TRUE), "\n")
  return(chunk)
}

# 自动适配文件流/标准输入流
input_stream <- if (opt$csv == "stdin") file("stdin") else opt$csv
read_csv_chunked(
  file = input_stream,
  callback = SideEffectChunkCallback$new(process_chunk),
  chunk_size = opt$chunk_size,
  show_col_types = FALSE
)

需要逐行处理时将chunk_size参数设为1即可。

方案2：base R原生实现（无第三方依赖） #

如果不想额外安装依赖包，可以用R原生的连接和读取函数实现流式处理，适合轻量部署场景：

library(optparse)

option_list <- list(
  make_option(c("-c", "--csv"), type = "character", default = "stdin",
              help = "CSV文件路径，默认读取标准输入流", metavar = "character"),
  make_option(c("-s", "--chunk-size"), type = "integer", default = 10000,
              help = "每批读取的行数，默认10000行", metavar = "integer")
)
opt_parser <- OptionParser(option_list = option_list)
opt <- parse_args(opt_parser)

# 建立输入流连接
con <- if (opt$csv == "stdin") {
  file("stdin", "r")
} else {
  file(opt$csv, "r")
}

# 读取CSV表头，无表头的场景可跳过该步骤自行指定列名
header <- strsplit(readLines(con, n = 1), ",")[[1]]

# 循环读取分块直到流结束
while(TRUE) {
  lines <- readLines(con, n = opt$chunk_size)
  if (length(lines) == 0) break
  # 解析当前分块为数据框
  chunk <- read.csv(text = lines, header = FALSE, col.names = header)
  # 在此处添加自定义处理逻辑
  print(paste0("当前处理块行数：", nrow(chunk)))
}

# 关闭连接释放资源
close(con)

注意事项 #

• CLI管道场景下不需要传入--csv参数，直接通过管道传输数据即可，示例用法：cat large_data.csv | Rscript your_script.R
• 处理gzip等压缩格式的流时，将连接替换为gzcon()即可直接读取，无需先解压到磁盘
• 特殊分隔符、引号转义等复杂CSV格式，优先使用readr方案，原生read.csv对异常格式的兼容性更差

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