foreach - R中SQLite数据库的并行查询

Question

我有一个大型数据库(~100Gb)，我需要从中提取每个条目， 对其进行一些比较，然后存储这些比较的结果。我尝试在单个 R session 中运行并行查询，但没有成功。我可以同时运行多个 R session ，但我正在寻找更好的方法。这是我尝试过的:

library(RSQLite)
library(data.table)
library(foreach)
library(doMC)



#---------
# SETUP
#---------


#connect to db
db <- dbConnect(SQLite(), dbname="genes_drug_combos.sqlite")


#---------
# QUERY
#---------
# 856086 combos = 1309 * 109 * 6

registerDoMC(8)

#I would run 6 seperate R sessions (one for each i)
res_list <- foreach(i=1:6) %dopar% {

  a <- i*109-108
  b <- i*109

  pb  <- txtProgressBar(min=a, max=b, style=3)
  res <- list()

  for (j in a:b) {

    #get preds for drug combos
    statement   <- paste("SELECT * from combo_tstats WHERE rowid BETWEEN", (j*1309)-1308, "AND", j*1309)
    combo_preds <- dbGetQuery(db, statement)

    #here I do some stuff to the result returned from the query
    combo_names <- combo_preds$drug_combo
    combo_preds <- as.data.frame(t(combo_preds[,-1]))

    colnames(combo_preds)  <- combo_names

    #get top drug combos
    top_combos <- get_top_drugs(query_genes, drug_info=combo_preds, es=T)

    #update progress and store result
    setTxtProgressBar(pb, j)
    res[[ length(res)+1 ]] <- top_combos
  }
  #bind results together
  res <- rbindlist(res)
}

我没有收到任何错误，但只有一个核心旋转。相反，如果我运行多个 R session ，我的所有核心都会执行它。我做错了什么？

Answer 1

我在使用RSQLite同时访问同一文件SQLite数据库时学到的一些东西:

1。确保每个工作人员都有自己的数据库连接。

  parallel::clusterEvalQ(cl = cl, {
    db.conn <- RSQLite::dbConnect(RSQLite::SQLite(), "./export/models.sqlite");
    RSQLite::dbClearResult(RSQLite::dbSendQuery(db.conn, "PRAGMA busy_timeout=5000;"));
  })

2。使用PRAGMA busy_timeout=5000;

默认情况下，此值设置为 0，并且每次您的工作线程在数据库锁定时尝试写入数据库时​​，您很可能会遇到“数据库已锁定”错误。前面的代码在每个工作连接中设置此PRAGMA。请注意，SELECT 操作永远不会被锁定，只有 INSERT/DELETE/UPDATE。

3。使用PRAGMAjournal_mode=WAL;

这只需要设置一次，并且默认情况下永远保持打开状态。它将向数据库添加两个(或多或少是永久性的)文件。它将 improve并发读/写性能。了解更多 here .

以上设置我没有经历过this问题。