R: accelerare le operazioni "gruppo per"

Question

Ho una simulazione che ha un enorme aggregato e combina il passo nel mezzo. Ho prototipato questo processo usando la funzione dDply () di Plyr che funziona alla grande per un'enorme percentuale delle mie esigenze. Ma ho bisogno di questo passaggio di aggregazione per essere più veloce poiché devo eseguire simulazioni 10k. Sto già ridimensionando le simulazioni in parallelo, ma se questo passo fosse più veloce potrei ridurre notevolmente il numero di nodi di cui ho bisogno.

Ecco una ragionevole semplificazione di ciò che sto cercando di fare:

library(Hmisc)

# Set up some example data
year <-    sample(1970:2008, 1e6, rep=T)
state <-   sample(1:50, 1e6, rep=T)
group1 <-  sample(1:6, 1e6, rep=T)
group2 <-  sample(1:3, 1e6, rep=T)
myFact <-  rnorm(100, 15, 1e6)
weights <- rnorm(1e6)
myDF <- data.frame(year, state, group1, group2, myFact, weights)

# this is the step I want to make faster
system.time(aggregateDF <- ddply(myDF, c("year", "state", "group1", "group2"),
                     function(df) wtd.mean(df$myFact, weights=df$weights)
                                 )
           )

Tutti i suggerimenti o i suggerimenti sono apprezzati!

Answer 1

Invece del normale frame di dati R, è possibile utilizzare un frame di dati immutabile che restituisce i puntatori all'originale quando si è sottolineato e può essere molto più veloce:

idf <- idata.frame(myDF)
system.time(aggregateDF <- ddply(idf, c("year", "state", "group1", "group2"),
   function(df) wtd.mean(df$myFact, weights=df$weights)))

#    user  system elapsed 
# 18.032   0.416  19.250 

Se dovessi scrivere una funzione Plyr personalizzata esattamente in questa situazione, farei qualcosa del genere:

system.time({
  ids <- id(myDF[c("year", "state", "group1", "group2")], drop = TRUE)
  data <- as.matrix(myDF[c("myFact", "weights")])
  indices <- plyr:::split_indices(seq_len(nrow(data)), ids, n = attr(ids, "n"))

  fun <- function(rows) {
    weighted.mean(data[rows, 1], data[rows, 2])
  }
  values <- vapply(indices, fun, numeric(1))

  labels <- myDF[match(seq_len(attr(ids, "n")), ids), 
    c("year", "state", "group1", "group2")]
  aggregateDF <- cbind(labels, values)
})

# user  system elapsed 
# 2.04    0.29    2.33 

È molto più veloce perché evita di copiare i dati, estraendo il sottoinsieme necessario solo per ciascun calcolo quando viene calcolato. La commutazione dei dati in forma di matrice fornisce un altro aumento della velocità perché il sottoinsieme della matrice è molto più veloce del sottoinsieme del frame di dati.

Answer 2

Ulteriori 2x speedUp e codice più conciso:

library(data.table)
dtb <- data.table(myDF, key="year,state,group1,group2")
system.time( 
  res <- dtb[, weighted.mean(myFact, weights), by=list(year, state, group1, group2)] 
)
#   user  system elapsed 
#  0.950   0.050   1.007 

Il mio primo post, quindi per favore sii gentile;)

---

Da data.table v1.9.2, setDT la funzione viene esportata che si convertirà data.frame a data.table come riferimento (in linea con data.table Parlanza - tutto set* funzioni Modificare l'oggetto per riferimento). Ciò significa che nessuna copia non necessaria ed è quindi veloce. Puoi cronometrare, ma sarà negligente.

require(data.table)
system.time({
  setDT(myDF)
  res <- myDF[, weighted.mean(myFact, weights), 
             by=list(year, state, group1, group2)] 
})
#   user  system elapsed 
#  0.970   0.024   1.015 

Questo è al contrario di 1,264 secondi con la soluzione di OP sopra, dove data.table(.) viene utilizzato per creare dtb.

Answer 3

Vorrei profilare con la base r

g <- with(myDF, paste(year, state, group1, group2))
x <- with(myDF, c(tapply(weights * myFact, g, sum) / tapply(weights, g, sum)))
aggregateDF <- myDF[match(names(x), g), c("year", "state", "group1", "group2")]
aggregateDF$V1 <- x

Sulla mia macchina ci vogliono 5sec rispetto a 67sec con il codice originale.

MODIFICAREHo appena trovato un'altra velocità con rowsum funzione:

g <- with(myDF, paste(year, state, group1, group2))
X <- with(myDF, rowsum(data.frame(a=weights*myFact, b=weights), g))
x <- X$a/X$b
aggregateDF2 <- myDF[match(rownames(X), g), c("year", "state", "group1", "group2")]
aggregateDF2$V1 <- x

Ci vogliono 3 secondi!

Answer 4

Stai usando l'ultima versione di Plyr (Nota: questo non è ancora arrivato a tutti gli specchi di cran)? In tal caso, potresti semplicemente eseguirlo in parallelo.

Ecco l'esempio llty, ma lo stesso dovrebbe applicarsi a ddply:

  x <- seq_len(20)
  wait <- function(i) Sys.sleep(0.1)
  system.time(llply(x, wait))
  #  user  system elapsed 
  # 0.007   0.005   2.005 

  library(doMC)
  registerDoMC(2) 
  system.time(llply(x, wait, .parallel = TRUE))
  #  user  system elapsed 
  # 0.020   0.011   1.038 

Modificare:

Bene, altri approcci di loop sono peggiori, quindi probabilmente richiede (a) codice C/C ++ o (b) un ripensamento più fondamentale di come lo stai facendo. Non ho nemmeno provato a usare by() Perché è molto lento nella mia esperienza.

groups <- unique(myDF[,c("year", "state", "group1", "group2")])
system.time(
aggregateDF <- do.call("rbind", lapply(1:nrow(groups), function(i) {
   df.tmp <- myDF[myDF$year==groups[i,"year"] & myDF$state==groups[i,"state"] & myDF$group1==groups[i,"group1"] & myDF$group2==groups[i,"group2"],]
   cbind(groups[i,], wtd.mean(df.tmp$myFact, weights=df.tmp$weights))
}))
)

aggregateDF <- data.frame()
system.time(
for(i in 1:nrow(groups)) {
   df.tmp <- myDF[myDF$year==groups[i,"year"] & myDF$state==groups[i,"state"] & myDF$group1==groups[i,"group1"] & myDF$group2==groups[i,"group2"],]
   aggregateDF <- rbind(aggregateDF, data.frame(cbind(groups[i,], wtd.mean(df.tmp$myFact, weights=df.tmp$weights))))
}
)

Answer 5

Di solito utilizzo un vettore indice con tocchetto quando la funzione da applicare ha args più vettoriali:

system.time(tapply(1:nrow(myDF), myDF[c('year', 'state', 'group1', 'group2')], function(s) weighted.mean(myDF$myFact[s], myDF$weights[s])))
# user  system elapsed 
# 1.36    0.08    1.44 

Uso un semplice wrapper che è equivalente ma nasconde il pasticcio:

tmapply(list(myDF$myFact, myDF$weights), myDF[c('year', 'state', 'group1', 'group2')], weighted.mean)

---

Modificato per includere TMapply per il commento di seguito:

tmapply = function(XS, INDEX, FUN, ..., simplify=T) {
  FUN = match.fun(FUN)
  if (!is.list(XS))
    XS = list(XS)
  tapply(1:length(XS[[1L]]), INDEX, function(s, ...)
    do.call(FUN, c(lapply(XS, `[`, s), list(...))), ..., simplify=simplify)
}