使用foreach和%dopar%并行计算KDN复杂度耗时更长的问题排查

并行代码的核心错误 #

你的并行实现存在迭代器使用错误：

• 串行代码中ix是1到n的索引，但并行代码里用iter(distance_matrix, by='row')得到的ix是距离矩阵的行数据向量，而非索引。这导致class_ix=dataY[ix]、distance_matrix[ix,]等操作完全偏离预期——你不是用索引取对应样本的类别，而是用距离值去索引dataY，这不仅逻辑错误，还会产生大量无效计算和内存开销，直接导致并行效率暴跌。

修复后的并行代码 #

先修正迭代器问题，同时优化集群变量导出，减少节点通信开销：

######################################
### 修正后的并行KDN计算 ###
######################################

## 准备并行环境
n.cores <- parallel::detectCores() - 1
my.cluster <- parallel::makeCluster(n.cores, type = "PSOCK")

# 显式导出必要变量到集群节点，避免自动复制冗余数据
parallel::clusterExport(my.cluster, 
                        varlist = c("dataY", "k_neighbours_classes", "distance_matrix"))
doParallel::registerDoParallel(cl = my.cluster)

kdn_instance <- NULL

system.time(
  kdn_instance <- foreach(
    ix = 1:n,  # 用索引迭代，而非行数据
    .combine = 'c',
    .packages = NULL  # 这里不需要额外加载包，提前导出变量即可
  ) %dopar% {
    class_ix <- dataY[ix]
    k_value <- k_neighbours_classes[class_ix]
    
    distances_ix <- distance_matrix[ix,]
    # 优化排序：只取前k_value+1个最小的索引，不用排序整个向量
    distances_ix_ordered <- order(distances_ix, decreasing = F)[1:(k_value+1)]
    knn_set_ix <- distances_ix_ordered[-1]  # 去掉自身（第一个元素是样本自己）
    
    Y_value_knn_set_ix <- dataY[knn_set_ix]
    Y_value_ix <- dataY[ix]
    
    knn_diff <- sum(Y_value_knn_set_ix != Y_value_ix)  # 用sum替代length+子集，更高效
    knn_diff / k_value
  }
)

parallel::stopCluster(cl = my.cluster)

更优的实现方式 #

1. 串行代码优化：用专用近邻计算包替代全矩阵排序 #

全距离矩阵的每行排序是串行耗时的核心，用FNN包的knn.index可以直接计算每个样本的k近邻索引，无需手动排序整个行向量，能大幅降低串行耗时：

library(FNN)

# 先按类别分组计算每个类的k值
class_k <- as.numeric(k_neighbours_classes[dataY])

# 计算每个样本的(class_k+1)近邻（包含自身）
knn_indices <- knn.index(dataX, k = max(class_k)+1)

# 逐个样本计算KDN
kdn_instance <- numeric(n)
system.time(
  for(ix in 1:n){
    k_val <- class_k[ix]
    # 取除自身外的k_val个近邻
    neighbours <- knn_indices[ix, 2:(k_val+1)]
    kdn_instance[ix] <- sum(dataY[neighbours] != dataY[ix]) / k_val
  }
)

这个优化能把串行耗时从12秒压缩到1-2秒左右。

2. 并行优化：结合分块计算减少通信开销 #

如果依然需要并行，建议按分块迭代（比如把1:n分成若干块，每个节点处理一块），减少foreach的迭代次数，降低节点通信的开销：

n.cores <- parallel::detectCores() - 1
my.cluster <- parallel::makeCluster(n.cores, type = "PSOCK")
parallel::clusterExport(my.cluster, varlist = c("dataY", "class_k", "knn_indices"))
doParallel::registerDoParallel(cl = my.cluster)

# 分块
chunks <- split(1:n, cut(1:n, n.cores, labels = FALSE))

system.time(
  kdn_list <- foreach(chunk = chunks) %dopar% {
    res <- numeric(length(chunk))
    for(i in seq_along(chunk)){
      ix <- chunk[i]
      k_val <- class_k[ix]
      neighbours <- knn_indices[ix, 2:(k_val+1)]
      res[i] <- sum(dataY[neighbours] != dataY[ix]) / k_val
    }
    res
  }
)
kdn_instance <- unlist(kdn_list)
parallel::stopCluster(cl = my.cluster)

是否需要更换包？ #

• 并行框架：foreach+doParallel本身没问题，但如果追求更简洁的并行语法，可以试试furrr（基于future的tidy风格并行），本质逻辑和上述方案一致。
• 近邻计算：强烈推荐FNN或RANN包，它们用C++实现近邻查找，比手动排序距离矩阵高效几个数量级。

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