RPostgreSQL 连接在使用 doParallel clusterEvalQ 启动后立即过期

Posted 2023-02-19

【中文标题】RPostgreSQL 连接在使用 doParallel clusterEvalQ 启动后立即过期

【英文标题】：RPostgreSQL connections are expired as soon as they are initiated with doParallel clusterEvalQ

【发布时间】：2015-09-11 07:18:07

【问题描述】：

我正在尝试设置一个并行任务，每个工作人员都需要进行数据库查询。我正在尝试为每个工作人员设置一个连接，如this question 中所示，但每次我尝试它都会返回<Expired PostgreSQLConnection:(2781,0)>，无论我注册了多少工作人员。

这是我的代码：

cl <- makeCluster(detectCores())
registerDoParallel(cl)

clusterEvalQ(cl, 
  library(RPostgreSQL)
  drv<-dbDriver("PostgreSQL")
  con<-dbConnect(drv, user="user", password="password", dbname="ISO",host="localhost")

)

如果我尝试运行我的foreach，尽管出现错误，它会以task 1 failed - "expired PostgreSQLConnection" 失败

当我进入 postgres 服务器状态时，它会显示所有已创建的活动会话。

我在与我的主 R 实例中的 postgres 交互时没有任何问题。

如果我跑步

clusterEvalQ(cl, 
  library(RPostgreSQL)
  drv<-dbDriver("PostgreSQL")
  con<-dbConnect(drv, user="user", password="password", dbname="ISO",host="localhost")
  dbGetQuery(con, "select inet_client_port()")

)

然后它将返回所有客户端端口。它不会给我过期通知，但如果我尝试运行我的 foreach 命令，它将失败并出现同样的错误。

编辑：

我已经在 Ubuntu 和 2 台 Windows 计算机上尝试过，它们都给出了相同的错误。

另一个编辑：

现在 3 台 windows 电脑

【问题讨论】：

您能否将您的 foreach 代码也添加到帖子中？

@JellenVermeir 无论foreach 中的其他内容如何，​​它都会在任何dbGetQuery 命令中失败。例如foreach(i=1:4) %dopar% dbGetQuery(con, "select * from sometable limit 1") 会失败，但foreach(i=1:4) %do% dbGetQuery(con, "select * from sometable limit 1") 不会失败。当我说它失败时，我的意思是我收到有关“过期 PostgreSQLConnection”的错误消息

【参考方案1】：

我能够在本地重现您的问题。我不完全确定，但我认为问题与clusterEvalQ 在内部的工作方式有关。比如你说dbGetQuery(con, "select inet_client_port()) 给你客户端端口输出。如果查询实际上是在集群节点上评估/执行的，那么您将无法看到此输出（与您无法直接读取在外部集群节点上执行的任何其他输出或打印语句相同的方式）。

因此，据我了解，评估首先在本地环境中执行，随后将相关函数和变量复制/导出到各个集群节点。这适用于任何其他类型的函数/变量，但显然不适用于数据库连接。如果连接/端口映射链接到主 R 实例，则连接将无法从从属实例工作。如果您尝试使用 clusterExport 函数来导出在主实例上创建的连接，您也会得到完全相同的错误。

作为替代方案，您可以在各个 foreach 任务中创建单独的连接。我已经用本地数据库验证了以下工作：

library(doParallel)
nrCores = detectCores()
cl <- makeCluster(nrCores)
registerDoParallel(cl)
clusterEvalQ(cl,library(RPostgreSQL))
clusterEvalQ(cl,library(DBI))

result <- foreach(i=1:nrCores) %dopar%

  drv <- dbDriver("PostgreSQL")
  con <- dbConnect(drv, user="user", password="password", dbname="ISO",host="localhost")
  queryResult <- dbGetQuery(con, "fetch something...")
  dbDisconnect(con)
  return(queryResult)

stopCluster(cl)

但是，现在您必须考虑到您将在每次 foreach 迭代时创建和断开新连接。因此，您可能会产生一些性能开销。您显然可以通过智能地拆分查询/数据来规避这种情况，以便在同一次迭代中完成大量工作。理想情况下，您应该将工作分配到您可用的尽可能多的内核中。