并行加载大量CSV文件至MySQL时如何避免超时错误

你的问题核心在于多个并行进程同时写入同一张MySQL表，当大文件写入时会触发表锁（尤其是MyISAM引擎，InnoDB虽然支持行锁，但批量插入也可能导致较长时间的锁占用），导致其他进程等待超时。既然你想在R层面解决，最稳妥的方案是把「数据预处理」和「数据库写入」解耦：并行做预处理，串行做写入——让worker只负责CPU密集的清洗工作，把处理好的数据交给主进程统一按顺序写入，这样既保留了并行预处理的效率，又彻底避免了锁表冲突。

下面是具体的实现步骤和代码修改建议：

1. 调整并行循环：只返回预处理结果，不在worker中写库 #

修改你的foreach循环，去掉worker里的数据库连接、写入逻辑，只返回处理好的业务数据和统计信息（空文件返回标记即可）。这样worker专注于数据处理，不碰IO操作：

library(foreach)
library(doParallel)

# 初始化并行集群（留一个核心给主进程做写入）
cl <- makeCluster(detectCores() - 1)
registerDoParallel(cl)

# 并行预处理：仅返回处理结果，不执行数据库操作
Bluetooth.processed <- foreach(i = jump:LoopLength, .inorder = TRUE, .packages = c("DBI", "RMariaDB")) %dopar% {
  Start.Time <- Sys.time()
  x <- filenames[i]
  
  if(!length(readLines(x))) {
    # 空文件：返回统计信息
    Stats <- data.frame(i = i, time = difftime(Sys.time(), Start.Time, units='secs'), DataList = 0)
    list(type = "empty", stats = Stats)
  } else {
    # 加载并清洗数据（保留你原有的逻辑）
    print(paste(i,"Start"))
    datalist <- readLines(x, encoding="UTF-8")
    datalist.length <- length(datalist)
    print(datalist.length)
    
    deviceID <- gsub("([0-9]+),.*", "\\1", datalist)
    Time <- gsub("[0-9]+,([0-9]+),.*", "\\1", datalist)
    Label <- gsub("^[0-9]+,[0-9]+,(.*),([a-zA-Z0-9:]+),([^,]+)$", "\\1", datalist)
    MacAddress <- gsub("^[0-9]+,[0-9]+,(.*),([a-zA-Z0-9:]+),([^,]+)$", "\\2", datalist)
    Strength <- gsub("^[0-9]+,[0-9]+,(.*),([a-zA-Z0-9:]+),([^,]+)$", "\\3", datalist)
    
    Label <- BlueToothFilterRules(Label)
    Encoding(Label) <- 'UTF-8'
    BlueToothTable <- data.frame(i = i, DeviceID = deviceID, Time = Time, Label = Label, Mac = MacAddress, Strength = Strength, stringsAsFactors = FALSE)
    Stats <- data.frame(i = i, time = difftime(Sys.time(), Start.Time, units='secs'), DataList = datalist.length)
    
    print(paste(i,"END"))
    list(type = "data", data = BlueToothTable, stats = Stats)
  }
}

# 关闭并行集群
stopCluster(cl)

2. 主进程串行写入数据库 #

在主进程中，建立一次数据库连接，然后遍历所有预处理结果，按顺序写入。这样所有写入操作都是串行的，不会有锁表冲突：

# 主进程建立数据库连接（仅连接一次，避免重复开销）
source("/Shared_Functions/MYSQL_LOGIN.r")
dbSendQuery(con, 'set character set "utf8"')

# 遍历预处理结果，串行执行写入
for(result in Bluetooth.processed) {
  if(result$type == "empty") {
    # 写入空文件统计数据
    invisible(dbWriteTable(con, name = "RawData_Loadloop_Stats_Bluetooth", value = result$stats, append = TRUE, row.names = FALSE))
  } else {
    # 写入业务数据 + 统计数据
    invisible(dbWriteTable(con, name = "RawData_Events_Bluetooth", value = result$data, append = TRUE, row.names = FALSE))
    invisible(dbWriteTable(con, name = "RawData_Loadloop_Stats_Bluetooth", value = result$stats, append = TRUE, row.names = FALSE))
  }
}

# 断开数据库连接
dbDisconnect(con)

3. 额外优化建议 #

• 切换至InnoDB引擎：如果你的MySQL表使用的是MyISAM，建议更换为InnoDB，它支持行级锁，即使后续有小批量并行写入需求，冲突概率也会大幅降低。
• 批量插入拆分：如果处理后的BlueToothTable数据量极大，可以使用dbWriteTable的batch_size参数（RMariaDB/RMySQL新版本支持），将大份数据拆分为若干小批次插入，减少单次锁表的持续时间。
• 内存分批控制：若1500个文件处理后的数据占用内存过高，可以改成「分批并行预处理 + 分批写入」模式：比如每处理50个文件就执行一次写入，再处理下一批，避免内存溢出。

这样修改后，你既保留了并行预处理的效率（其他worker可在主进程写入时继续处理下一批文件），又彻底解决了锁表超时问题，且代码无需修改MySQL配置，在不同机器上都能直接运行。

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