嵌套循环中使用multidplyr出现“closing unused connections”警告原因排查

我正在运行嵌套循环，对大型稀疏数据集做子集划分，还要针对2年、3年、5年的搜索窗口，计算每个实验室变量的均值和中位数。现在出现warning: closing unused connections警告，想问是不是因为multidplyr在循环里没法用其他核心导致的？

我的代码逻辑（对应截图内容） #

library(multidplyr)
# 创建4核集群并加载依赖包
cluster <- create_cluster(4) %>% 
  cluster_library(c("dplyr", "lubridate"))

# 嵌套循环处理不同窗口和变量
for (window in c(2, 3, 5)) {
  for (var in lab_vars) {
    df_sub <- df %>% 
      # 数据集子集划分逻辑
      filter(...) %>% 
      # 按ID分区到集群
      partition(ID, cluster = cluster) %>% 
      # 基于当前窗口过滤时间范围
      mutate(date_diff = interval(start_date, end_date) %>% as.numeric("years")) %>% 
      filter(date_diff <= window) %>% 
      # 计算均值和中位数
      summarise(
        mean_val = mean(!!sym(var), na.rm = TRUE),
        median_val = median(!!sym(var), na.rm = TRUE)
      ) %>% 
      # 把结果拉回本地
      collect()
    
    # 结果保存逻辑（比如写入文件或存入列表）
  }
}

问题解答 #

这个警告不是因为multidplyr没法在循环内使用多核，核心原因是循环里反复调用partition()创建新分区，但旧的分区连接没被主动清理，R会自动关闭这些闲置连接并抛出警告。

每次partition()都会在集群节点上创建新的连接，循环反复执行这个操作后，大量闲置连接堆积，R就会触发这个警告来提示你它在清理这些没用的连接。

解决办法 #

方法1：提前分区，避免循环内重复创建连接（推荐）

把数据集的分区操作放到循环外面，循环内直接复用已分区的数据集，这样只会创建一次连接，从根源避免警告：

library(multidplyr)
cluster <- create_cluster(4) %>% 
  cluster_library(c("dplyr", "lubridate"))

# 提前按ID完成分区，放在循环外
df_partitioned <- df %>% partition(ID, cluster = cluster)

for (window in c(2, 3, 5)) {
  for (var in lab_vars) {
    result <- df_partitioned %>% 
      mutate(date_diff = interval(start_date, end_date) %>% as.numeric("years")) %>% 
      filter(date_diff <= window) %>% 
      summarise(
        mean_val = mean(!!sym(var), na.rm = TRUE),
        median_val = median(!!sym(var), na.rm = TRUE)
      ) %>% 
      collect()
    
    # 保存结果
  }
}

# 循环结束后务必关闭集群释放资源
stop_cluster(cluster)

方法2：循环内主动清理闲置连接

如果必须在循环内做子集划分和分区，那每次处理完结果后，主动删除分区对象并触发垃圾回收，让R及时清理连接：

library(multidplyr)
cluster <- create_cluster(4) %>% 
  cluster_library(c("dplyr", "lubridate"))

for (window in c(2, 3, 5)) {
  for (var in lab_vars) {
    df_sub <- df %>% 
      filter(...) %>% 
      partition(ID, cluster = cluster) %>% 
      mutate(date_diff = interval(start_date, end_date) %>% as.numeric("years")) %>% 
      filter(date_diff <= window) %>% 
      summarise(
        mean_val = mean(!!sym(var), na.rm = TRUE),
        median_val = median(!!sym(var), na.rm = TRUE)
      ) %>% 
      collect()
    
    # 保存结果
    
    # 主动清理分区对象和连接
    rm(df_sub)
    gc()
  }
}

stop_cluster(cluster)

补充说明 #

这个警告本身不会影响计算结果，但会占用额外的系统资源，处理后既能消除警告，也能让代码运行更高效。

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