R 中的并行化

R 并行编程

Nabeel Imam

Data Scientist

实用示例

数据

print(file_list)

 [1] "./uni_data_country/Argentina.csv"
 [2] "./uni_data_country/Armenia.csv"
 [3] "./uni_data_country/Australia.csv"
 [4] "./uni_data_country/Austria.csv"
 [5] "./uni_data_country/Azerbaijan.csv"
 [6] "./uni_data_country/Bahrain.csv"
 [7] "./uni_data_country/Bangladesh.csv"
 [8] "./uni_data_country/Belarus.csv"
 [9] "./uni_data_country/Belgium.csv"
[10] "./uni_data_country/Bolivia.csv"
...

显示三栋大学建筑与毕业生，并标注一至三名的排名。

添加一列

for (file in file_list) {

  df <- read.csv(file)


  df$top100 <- NA

  for (r in 1:nrow(df)) {
    df$top100[r] <- df$world_rank[r] <= 100
  }


  write.csv(df, file)
}

性能分析

代码

library(profvis)


profvis({

  for (file in file_list) {

    df <- read.csv(file)
    df$top100 <- NA

    for (r in 1:nrow(df)) {
      df$top100[r] <- df$Rank[r] <= 100
    }
    write.csv(df, file)
  }

})

输出

来自 profvis() 的性能分析输出。示例代码中，读取数据耗时 40 毫秒，填充 `top100` 列耗时 80 毫秒，其他步骤几乎瞬时完成。

让我们并行化

循环

  for (file in file_list) {

    df <- read.csv(file)
    df$top100 <- NA

    for (r in 1:nrow(df)) {
      df$top100[r] <- df$Rank[r] <= 100
    }
    write.csv(df, file)
  }

函数

add_col <- function(file_path) {

  df <- read.csv(file_path)
  df$top100 <- NA

  for (r in 1:nrow(df)) {
    df$top100[r] <- df$Rank[r] <= 100
  }
  write.csv(df, file_path)
}


cl <- makeCluster(6)

dummy <- parLapply(cl, file_list, add_col)
stopCluster(cl)

实务考量：内核数量

检测内核数

detectCores()

[1] 8

并行代码

cl <- makeCluster(detectCores() - 2)


dummy <- parLapply(cl, file_list, add_col)

stopCluster(cl)

实务考量：集群类型

PSOCK 集群（默认）

cl <- makeCluster(detectCores() - 2)

创建当前 R 会话的副本
内核不共享内存
适用于所有操作系统（Windows、Mac、Linux）

FORK 集群

cl <- makeCluster(detectCores() - 2,
                  type = "FORK")

从 R 会话创建子进程
内核共享内存（比 PSOCK 更快）
不支持 Windows

让我们练习！

R 并行编程