paralleles Lesen und Verarbeiten von Dateien in R

Question

Ich verwende die Bibliothek parallel in R, um einen großen Datensatz zu verarbeiten, auf den ich komplexe Operationen anwende.

Aus Gründen einen reproduzierbaren Code bereitzustellen, können Sie unter einem einfacheren Beispiel finden:

#data generation 
dir <- "C:/Users/things_to_process/" 

setwd(dir) 
for(i in 1:800) 
{ 
    my.matrix <- matrix(runif(100),ncol=10,nrow=10) 

    saveRDS(my.matrix,file=paste0(dir,"/matrix",i)) 
} 

#worker function 
worker.function <- function(files) 
{ 
    files.length <- length(files) 
    partial.results <- vector('list',files.length) 

    for(i in 1:files.length) 
    { 
     matrix <- readRDS(files[i]) 
     partial.results[[i]] <- sum(diag(matrix)) 
    } 

    Reduce('+',partial.results) 
} 


#master part 
cl <- makeCluster(detectCores(), type = "PSOCK") 

file_list <- list.files(path=dir,recursive=FALSE,full.names=TRUE) 

part <- clusterSplit(cl,seq_along(file_list)) 
files.partitioned <- lapply(part,function(p) file_list[p]) 

results <- clusterApply(cl,files.partitioned,worker.function) 

result <- Reduce('+',results) 

Wesentlichen statt in einer verschachtelten Weise geschehen würde, wenn versucht, ich frage mich, Dateien parallel zu lesen. Und wenn dieser Engpass die erwartete Leistung parallel laufender Aufgaben reduzieren würde?

Wäre es besser, wenn ich zuerst alle Matrizen auf einmal in einer Liste lese und dann Teile dieser Liste an jeden Kern sende, damit er verarbeitet werden kann? Was wäre, wenn diese Matrizen viel größer wären, wäre ich in der Lage, alle in einer Liste auf einmal zu laden?

Answer 1

Anstatt die matrix in einer separaten RDS-Datei zu speichern, haben Sie versucht, eine list von N Matrizen in jeder Datei zu speichern, wobei N die Nummer ist, die von einem einzelnen Arbeiter verarbeitet wird?

Dann sieht die worker.function wie:

worker.function <- function(file) { 
    matrix_list <- readRDS(file) 
    partial_results <- lapply(matrix_list, function(mat) sum(diag(mat))) 
    Reduce('+',partial.results) 
} 

Sie sollten einige Zeit auf der I/O sparen und vielleicht sogar auf die Berechnung durch einen for mit einem lapply ersetzen.