R语言处理大型固定宽文件：定位特定表头并提取数据

问题概述 #

需要处理单文件3GB+的固定宽度文本，每个文件包含Sensor Messages、Tag Records、Setup Messages三类数据：

• 每类数据有专属标记行（如Sensor Messages:），随后是列头，列头与数据间用等号线分隔
• 需求：遍历文件夹内所有文件，提取三类数据并添加原文件名，分别导出为CSV，其中Tag Records优先级最高
• 原方案用readr::read_fwf()时，全文件扫描定位表头效率极低，且无法快速传递定位结果读取目标数据

核心解决思路 #

避免全量读取大文件，通过逐行扫描定位数据块边界，仅读取目标数据段：

1. 用base R的文件连接逐行读取，定位每类数据的起始行（等号线的下一行）和结束行（下一类数据标记的前一行）
2. 利用read_fwf()的skip和n_max参数，精准读取指定范围的固定宽度数据
3. 批量处理文件夹内所有文件，合并后导出

实现代码 #

1. 全局参数定义 #

library(readr)
library(dplyr)

# 各数据类型的列宽与列名
SensorWidths <- c(8, 10, 7, 9, 6, 10, 12, 6, 8, 8)
TagWidths <- c(8, 10, 12, 13, 13, 12, 9)
SetupWidths <- c(8, 10, 21, 47)

SensorCols <- c("Date", "Time", "Sensor", "Blank", "Temp", "Pressure", "Battery", "Tilt_X", "Tilt_Y", "Tilt_Z")
TagCols <- c("Date", "Time", "SubSec", "TagID", "TagType", "Sensor", "Power")
SetupCols <- c("Date", "Time", "Type", "Details")

# 数据类型标记（用于定位）
data_markers <- c("Sensor Messages:", "Tag Records:", "Setup Messages:")

2. 单文件处理函数 #

process_single_file <- function(file_path) {
  file_con <- file(file_path, "r")
  lines <- c()
  line_num <- 0
  block_positions <- list()
  
  # 逐行扫描，定位各数据块的起始/结束行
  while (length(current_line <- readLines(file_con, n = 1)) > 0) {
    line_num <- line_num + 1
    current_line_trim <- trimws(current_line)
    
    # 匹配数据类型标记
    marker_match <- match(current_line_trim, data_markers)
    if (!is.na(marker_match)) {
      marker_name <- gsub(":", "", data_markers[marker_match])
      # 跳过列头行（标记行+1），找到等号线行（标记行+2），数据起始行=等号线行+1
      header_line <- readLines(file_con, n = 1)
      line_num <- line_num + 1
      equal_line <- readLines(file_con, n = 1)
      line_num <- line_num + 1
      block_start <- line_num + 1
      
      # 记录当前块的起始位置和类型
      block_positions[[marker_name]] <- list(start = block_start, type = marker_name)
      
      # 如果是前一个块，补全结束位置
      if (length(block_positions) > 1) {
        prev_block <- names(block_positions)[length(block_positions)-1]
        block_positions[[prev_block]]$end <- line_num
      }
    }
  }
  # 补全最后一个块的结束位置（文件末尾）
  last_block <- names(block_positions)[length(block_positions)]
  block_positions[[last_block]]$end <- line_num
  
  close(file_con)
  file_name <- basename(file_path)
  
  # 读取各数据块
  sensor_data <- NULL
  tag_data <- NULL
  setup_data <- NULL
  
  # 优先处理Tag Records（优先级最高）
  if ("Tag Records" %in% names(block_positions)) {
    tag_block <- block_positions[["Tag Records"]]
    n_rows <- tag_block$end - tag_block$start + 1
    tag_data <- read_fwf(file_path, 
                         fwf_widths(TagWidths, TagCols),
                         skip = tag_block$start - 1,
                         n_max = n_rows) %>%
      mutate(Source_File = file_name)
  }
  
  if ("Sensor Messages" %in% names(block_positions)) {
    sensor_block <- block_positions[["Sensor Messages"]]
    n_rows <- sensor_block$end - sensor_block$start + 1
    sensor_data <- read_fwf(file_path, 
                            fwf_widths(SensorWidths, SensorCols),
                            skip = sensor_block$start - 1,
                            n_max = n_rows) %>%
      mutate(Source_File = file_name)
  }
  
  if ("Setup Messages" %in% names(block_positions)) {
    setup_block <- block_positions[["Setup Messages"]]
    n_rows <- setup_block$end - setup_block$start + 1
    setup_data <- read_fwf(file_path, 
                           fwf_widths(SetupWidths, SetupCols),
                           skip = setup_block$start - 1,
                           n_max = n_rows) %>%
      mutate(Source_File = file_name)
  }
  
  return(list(Sensor = sensor_data, Tag = tag_data, Setup = setup_data))
}

3. 批量处理与导出 #

# 指定目标文件夹
target_dir <- "path/to/your/files"
file_list <- list.files(target_dir, pattern = "\\.txt$", full.names = TRUE)

# 批量处理所有文件
all_results <- lapply(file_list, process_single_file)

# 合并各类型数据
combined_sensor <- bind_rows(lapply(all_results, function(x) x$Sensor))
combined_tag <- bind_rows(lapply(all_results, function(x) x$Tag))
combined_setup <- bind_rows(lapply(all_results, function(x) x$Setup))

# 导出为CSV
write_csv(combined_tag, "Tag_Records_Combined.csv")
write_csv(combined_sensor, "Sensor_Messages_Combined.csv")
write_csv(combined_setup, "Setup_Messages_Combined.csv")

关键说明 #

• 高效定位：仅逐行扫描文件的标记行和边界，无需全量加载大文件，内存占用极低
• 精准读取：通过skip和n_max参数直接读取目标数据段，避免无效数据处理
• 优先级保障：代码中优先处理Tag Records，确保其数据完整性优先于其他类型
• 兼容性：适配不同文件中数据块顺序变化的情况，无需依赖固定行号

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