Использовать параллельную обработку (parLapply) циклы (чтение и сохранение) по различным каталогам на сервере - PullRequest
Новая
       20

Использовать параллельную обработку (parLapply) циклы (чтение и сохранение) по различным каталогам на сервере

0 голосов
/ 06 июня 2019

Я новичок в параллельной обработке в R и пробовал несколько вариантов для запуска большой модели. Во-первых, эта модель не может работать на моем собственном ПК из-за недостатка памяти (я полагаю, это ОЗУ) для продолжения процесса.

Итак, я попытался сделать это на сервере с более высокой памятью, но по какой-то причине способ установки кода не сработал.

ЧТО МНЕ НУЖЕН, ЧТО ЭТОТ КОДЕКС ДЕЛАТЬ В ПРОСТЫХ СЛОВАХ: Мне нужно запустить мой model в 29 штатах США, которые хранятся в разных каталогах (например, E:/server/states/ar, E:/server/states/ca, E:/server/states/pa , так далее). Каждое состояние содержит 30 файлов с одинаковыми именами (то есть file1, file2, file3, ..., file30). Эти файлы являются входными данными, которые моя модель использует для своей цели и сохраняет выходные данные во фрейме данных, который я позже сохраняю как CSV.

Мне удалось адаптировать мой код для запуска с использованием параллельной обработки (parLapply), но только для одного состояния. Без параллельной обработки запуск модели для одного состояния занимает около 10 часов. При параллельной обработке (parLapply) это занимает чуть больше часа.

ЧТО Я ПОПРОБОВАЛ И ЧТО Я ПОЛУЧИЛ, ЧТОБЫ ПОЛУЧИТЬ: Я ищу способ автоматизировать этот процесс, просматривая различные каталоги, читая информацию о каждом состоянии, используя эту информацию для запуска моей модели и сохраняя вывод как csv в каталоге, который просто использовался для чтения входных данных (хотя на самом деле не имеет значения, где я сохраняю выходные данные, если они называются по-разному).

Я попробовал несколько комбинаций следующего кода, но, похоже, есть проблема: мне удалось пройтись по каталогам, но я не смог прочитать файлы из этих каталогов, а только из первого рабочего каталога.

Это очень странно, потому что мне удалось сохранить выходные данные в разных каталогах, поэтому я знаю, что это не проблема циклического перемещения по каталогам (на самом деле я подозреваю, что это связано с тем, где определяется параллельный кластер). Кстати, выходные данные, которые хранились в разных каталогах, были такими же, как и 30 файлов (входных данных) только из первого каталога в dirs1. Это означает, что код читает только 30 файлов из первого каталога и ничего больше.

Любые идеи здесь будут высоко оценены. 

library(data.table); library(lubridate); library(dplyr)
library(doSNOW); library(lme4); library(doParallel); library(parallel)

dirs <- list.dirs("E:/server/states")
dirs1 <- dirs[c(1,3,5,7,9,11,13,15,17,19,21,23,25,27,29,31,33,35,37,39)]

# Set working directory
setwd(dirs1[1])
getwd()

# Big model
model <- function(first, second, third, number) {
   # do all this stuff
   # apparently consumes a lot of RAM memory
   return (df)
}

# File names
filename <- c("file1", "file2", "file3" , "file4", "file5",
              "file6", "file7", "file8" , "file9", "file10",
              "file11", "file12", "file13" , "file14", "file15",
              "file16", "file17", "file18" , "file19", "file20",
              "file21", "file22", "file23" , "file24", "file25",
              "file26", "file27", "file28" , "file29", "file30")

names_states <- c("al", "ak", "ar", "az", "ca", "co", "de", "fl", "ga",
 "il", "in", "ky", "md", "mi", "mn", "mo", "ms", "mt", "ne", "nc", "nj", "ny", 
 "oh", "pa", "sc", "tn", "va", "wi", "wv")

## -------------- Parallel processing using parLapply --------------###

read.dirs <- lapply(1:20, function(m) {
  # Set working directory
  setwd(dirs1[m])
  getwd()

  # Initiate cluster
  cl <- makeCluster(detectCores()-2)

  registerDoParallel(cl)
  registerDoSNOW(cl)
  clusterEvalQ(cl, library(dplyr))
  clusterEvalQ(cl, library(lubridate))
  clusterEvalQ(cl, library(data.table))
  clusterExport(cl, list('model', 'filename', 'name_states', 'dirs', 'dirs1'))

  #Running cluster
  df.br <- parLapply(cl, 1:length(filename), function(i){
    txt <- filename[i]
    a <- paste("X",i, sep="")
    a <- model(first=45, second=50, third=txt, number=i)
    df <- rbind(a)
  })

  #Stop cluster after finish
  stopCluster(cl)

  df.result <- do.call(rbind, df.br)
  filen <- paste(names_states[m], "_", Sys.Date(),"_parallel.csv", sep="")
  write.csv(df.results, file=filen)

  # rm(list=ls())
  # gc()

})
...