有没有办法通过输入文件并行运行 R 脚本？答案

【问题标题】：Is there a way to parallelize running R script by input files?有没有办法通过输入文件并行运行 R 脚本？
【发布时间】：2021-12-06 11:58:58
【问题描述】：

我正在读取一堆文件（超过七千个）作为数据框。所有文件都在同一个父文件夹中，具有组织和一致的子目录。这些文件当前按时间戳组织。我想读取文件，然后将文件导出到不同的文件夹，其中每个文件都是一个玩家 ID。在同一输入数据帧中，单个玩家可能有多个时间戳。有时玩家 ID 根本不会出现在输入数据框中。我已经弄清楚了数据争吵（这很简单），但是由于每个文件大约有 150 万行，因此单个文件大约需要 5 个小时。所以我不能简单地遍历所有七千个文件。我想改为通过输入文件进行并行化（尽管通过输出文件进行并行化可能会更好？）。我将在具有足够 CPU 的 HPC 上运行它，并且在使用 HPC 之前我不需要指定我的 CPU 要求。我知道doParallel包是存在的，但是教程介绍vignette("gettingstartedParallel")没用，其他doParallel帖子我也看不懂。（请不要只向我介绍没有相关代码的 doParallel 包。）我还担心代码崩溃，因为它试图多次写入同一个 csv。即使我设置了append = TRUE，也无法并行写入 CSV。这是我如何读取文件以及如何将文件写入新文件夹的代码。

# Example input data frames (in the real code I create a vector of Alltimes using list.files() )
times1 <- data.frame(
  ID = c('PL1', 'PL2', 'PL3', 'PL2','PL1'),
  times = c(42.6, 41.5, 42.9, 47.0, 44.3),
  speed = c(64, 66, 43, 39, 55) 
)

times2 <- data.frame(
  ID = c('PL3', 'PL3', 'PL3', 'PL1','PL1'),
  times   = c(62.1, 51.7, 65.9, 62.1, 55.3),
  speed = c(71, 73, 45, 64, 66) 
)

# Create vector of all parquets filepaths
Alltimes <- list.files(path = 'Input_Folder_Path)',
                       pattern = '*.snappy.parquet$',
                       recursive = TRUE,
                       full.names = TRUE)

# Iterate through timestamp input files (I want this part parallelized instead of a loop)
# for( i in 1:length(Alltimes)){
  
  # Read in the individual file
  # when the Alltimes vector is the file path I use read_parquet( Alltimes[i] ), but
  # times1 is a substitute for this example.
  df = times1 
  # df = times2
  # df = read_parquet( Alltimes[i] )
  
  # get vector of all player ids in this data frame
  all_ids_vec <- unique(x = df$ID)
  
  # write out individual csv for each player ID
  for(j in 1:length(all_ids_vec)){
    
    # Subset the df by that specific player ID
    one_player <- df %>% filter(ID == all_ids_vec[j])
    
    write.table(x = one_player, 
                file = "C:/Users/Juliet/Desktop/", all_ids_vec[j],".csv", 
                append = TRUE, 
                quote = FALSE, 
                sep = ",", 
                row.names = FALSE, 
                col.names = FALSE)
  }
# }

【问题讨论】：

如果"每个文件都是一个玩家ID"那为什么unique(x = df$ID)?
我希望每个OutputFile.csv 都是玩家ID。每个InputFile.csv 中都有许多不同的玩家ID
好的，所以最好 split 由 ID 一次处理每个。现在我不明白的是，为什么处理 150 万行需要 5 小时，大部分时间肯定是在进行数据整理，对吧？
数据处理进行得非常快。我认为写出分离 csvs 需要很长时间。每个文件有约 150 万行和约 50 万个唯一 ID。所以我正在写出〜500K csvs，每个输入文件都有3行（这些数字是近似的，并且在输入文件中不一致。尽管每个输入文件的行数大致相同）。我认为这需要很长时间，因为我的 RAM 用完了，我正在本地计算机上进行测试。但是当我处理所有内容时，它将在 HPC 上，我可以分配更多 RAM。 split ID 的代码是什么？
按id分割的代码是df_split <- split(df, df$ID)。然后您可以使用lapply 处理它们中的每一个。这比一个一个过滤，分配给一个新的data.frame并保存它要快。您可以完全避免此过滤器和分配。

标签： r csv parallel-processing data-wrangling doparallel

【解决方案1】：

为了将文件并行写入磁盘，以下代码适用于 Windows。请注意，包 doParallel 不适用于 Windows。

library(parallel)

dirname <- "C:/Users/611913/Desktop"
#dirname <- path.expand("~/tmp/so")

sp <- split(df, df$ID)
id_vec <- names(sp)

f <- function(X, filename, path){
  filename <- file.path(path, filename)
  write.table(x = X, 
              file = filename, 
              append = TRUE, 
              quote = FALSE, 
              sep = ",", 
              row.names = FALSE, 
              col.names = FALSE)  
}

# Windows
ncores <- detectCores()

cl <- makeCluster(ncores - 1L)
clusterExport(cl, "sp")
clusterExport(cl, "id_vec")
clusterExport(cl, "dirname")
clusterExport(cl, "f")

clusterEvalQ(cl, "sp")
clusterEvalQ(cl, "id_vec")
clusterEvalQ(cl, "dirname")
clusterEvalQ(cl, "f")

res <- parLapply(cl, seq_along(sp), function(j){
  f(sp[[j]], id_vec[j], dirname)
})

stopCluster(cl)

测试数据

此代码创建测试数据。

n <- 1000L
ID <- sprintf("id%04d", seq_len(n))
ID <- rep(ID, 3)
df <- data.frame(ID, x = rnorm(3*n), y = sample(10, 3*n, TRUE))

【讨论】：