Windows中的并行for循环

Question

我对很远的行业级数据库进行了很多SQL查询,但接收结果需要很长时间.当我带R的计算机几乎紧挨着数据库时,速度要快得多,这让我相信它是我的计算机和数据库之间的延迟,这是瓶颈,并且运行并行查询可能会加快速度.我们在不同的大陆.

这是一个不平行的工作版本:

doQueries <- function(filenameX, inp1, inp2) {
  print(paste("Starting:", inp1, inp2, ",saving to", filenameX, sep=" "))
  # Here should the query be (using RODBC)
  # save(queryresults, file="filenameX")
}

input.rows <- cbind(c("file1.rda","file2.rda","file3.rda"),c("A","B","C"),c(12,13,14))

for (i in 1:nrow(input.rows)) {
  doQueries(filenameX=input.rows[i,1], inp1=input.rows[i,2], inp2=input.rows[i,3])
}

我已尝试使用以下代码,但foreach-library似乎并不可用,正如我从阅读CRAN时所理解的那样,并行正在替换早期的并行软件包("package'foreach'不可用(对于R版本2.15). 0)").

library(parallel)
library(foreach)
foreach (i=1:nrow(input.rows)) %dopar% {
  doQueries(filenameX=input.rows[i,1], inp1=input.rows[i,2], inp2=input.rows[i,3])
}

我应该怎么做呢？

感谢Stackoverflow上的所有贡献者!

/克里斯

更新:感谢nograpes,我设法加载库.以下代码似乎有效:

library(RODBC)
library(doParallel)
library(foreach)

# odbcCloseAll()
# my_conn <- odbcConnect("database", uid="xx", pwd="yy", case="nochange")

doQueries <- function(filenameX, inp1, inp2) {
  print(paste("Starting:", inp1, inp2, ",saving to", filenameX, sep=" "))
  # sql.test <- RODBC::sqlQuery(my_conn, "SELECT * FROM zzz LIMIT 100", rows_at_time=1024)
  # save(sql.test, file="filenameX")
}

input.rows <- cbind(c("file1.rda","file2.rda","file3.rda"),c("A","B","C"),c(12,13,14))

cl <- makeCluster(3)
registerDoParallel(cl)

foreach (i=1:nrow(input.rows)) %dopar% {
  doQueries(filenameX=input.rows[i,1], inp1=input.rows[i,2], inp2=input.rows[i,3])
}

stopCluster(cl)

但是当我包含实际的SQL查询时,会出现此错误消息:{:task 1 failed - "first argument is not open RODBC channel"中的错误

可能是这样,这在概念上是行不通的吗？那个RODBC一次不能处理多个查询？

我非常感谢所有的支持.

/克里斯

更新2:非常感谢nograpes非常好的和令人印象深刻的答案.很难判断数据传输本身是否更快(我认为总吞吐量快20%),但我发现,因为查询(大约100)具有不同的响应时间,并且需要后处理(我在函数中包含)在保存之前),我可以更好地利用链接和本地CPU.即,当时只有一个查询,在数据传输期间CPU几乎不用,然后在CPU工作时链路将保持安静.通过并行查询,我看到数据到达并且CPU同时工作.总的来说它变得更快.非常感谢!

/克里斯

Answer 1

正如我在评论中提到的,这种技术可能不会更快.要回答您的问题,该foreach软件包适用于您的R版本.可能您选择的存储库尚未更新.试试这个:

install.packages('foreach', repos='http://cran.us.r-project.org')

哪个应该安装包.如果这不起作用,请在此处获取操作系统的二进制文件,然后通过菜单安装它.

如果瓶颈是网络连接,那么您只能通过减少放在网络上的内容来加快流程.一个想法是远程连接到数据库服务器,让它将查询转储到文件(在服务器上),压缩它,然后将其下载到您的计算机,然后让R解压缩并加载它.这听起来很多,但你可以在R内完成整个过程.

跟进您的更新后,.packages您的foreach语句中似乎没有包含参数.这就是你必须为sqlQuery函数添加前缀的原因RODBC::.有必要指定循环所需的包,因为我认为它实际上为每个节点启动了一个新的R会话,并且每个会话都需要使用包进行初始化.同样,您无法访问,my_conn因为它位于循环之外,您需要在循环内部创建,以便每个节点都有自己的副本.

library(RODBC)
library(foreach)
library(doParallel)
setwd('C:/Users/x/Desktop')

doQueries <- function(filenameX) {
  sql.text<-sqlQuery(my_conn, 'SELECT * FROM table;')
  save(sql.text, file=filenameX)
}

cl <- makeCluster(2)
registerDoParallel(cl)

foreach (i=1:2, .packages='RODBC') %dopar% {
  my_conn <- odbcConnect("db", uid="user", pwd="pass")
  doQueries(filenameX=paste('file_',i,sep=''))
}

但是,正如我所提到的,这可能不会更快.