data.table y computación paralela

parallel-processing

Question

A continuación de esta publicación: tabla multinúcleo y tabla de datos en R , me preguntaba si habría una manera de usar todos los núcleos cuando se usa la tabla de datos, que normalmente los cálculos por grupos podrían ser paralelizados. Parece que plyr permite tales operaciones por diseño.

Answer 1

He hecho algunas pruebas según el mantra anterior de @matt dowle de Rprof, Rprof, Rprof.

Lo que encuentro es que la decisión de paralelizar depende del contexto; pero es probable que sea significativo. Dependiendo de las operaciones de prueba (por ejemplo, foo continuación, que se pueden personalizar) y la cantidad de núcleos utilizados (intento 8 y 24), obtengo resultados diferentes.

A continuación los resultados:

1. Usando 8 núcleos, veo una mejora del 21% en este ejemplo para la paralelización
2. Utilizando 24 núcleos, veo un 14% de mejora .

También observo algunos datos / operaciones del mundo real (no compartibles) que muestran una mejora de paralelización de mayor ( 33% o 25% , dos pruebas diferentes) con 24 núcleos. Edición de mayo de 2018 Un nuevo conjunto de casos de ejemplo del mundo real muestra mejoras cercanas al 85% de las operaciones paralelas con 1000 grupos.

R> sessionInfo() # 24 core machine:
R version 3.3.2 (2016-10-31)
Platform: x86_64-pc-linux-gnu (64-bit)
Running under: CentOS Linux 7 (Core)
attached base packages:
[1] parallel stats graphics grDevices utils datasets methods
[8] base
other attached packages:
[1] microbenchmark_1.4-2.1 stringi_1.1.2 data.table_1.10.4
R> sessionInfo() # 8 core machine:
R version 3.3.2 (2016-10-31)
Platform: x86_64-apple-darwin13.4.0 (64-bit)
Running under: macOS Sierra 10.12.4
attached base packages:
[1] parallel stats graphics grDevices utils datasets methods base
other attached packages:
[1] microbenchmark_1.4-2.1 stringi_1.1.5 data.table_1.10.4

Ejemplo a continuación:

library(data.table)
library(stringi)
library(microbenchmark)
set.seed(7623452L)
my_grps <- stringi::stri_rand_strings(n= 5000, length= 10)
my_mat <- matrix(rnorm(1e5), ncol= 20)
dt <- data.table(grps= rep(my_grps, each= 20), my_mat)
foo <- function(dt) {
dt2 <- dt ## needed for .SD lock
nr <- nrow(dt2)
idx <- sample.int(nr, 1, replace=FALSE)
dt2[idx,][, `:=` (
new_var1= V1 / V2,
new_var2= V4 * V3 / V10,
new_var3= sum(V12),
new_var4= ifelse(V10 > 0, V11 / V13, 1),
new_var5= ifelse(V9 < 0, V8 / V18, 1)
)]
return(dt2[idx,])
}
split_df <- function(d, var) {
base::split(d, get(var, as.environment(d)))
}
foo2 <- function(dt) {
dt2 <- split_df(dt, "grps")
require(parallel)
cl <- parallel::makeCluster(min(nrow(dt), parallel::detectCores()))
clusterExport(cl, varlist= "foo")
clusterExport(cl, varlist= "dt2", envir = environment())
clusterEvalQ(cl, library("data.table"))
dt2 <- parallel::parLapply(cl, X= dt2, fun= foo)
parallel::stopCluster(cl)
return(rbindlist(dt2))
}
print(parallel::detectCores()) # 8
microbenchmark(
serial= dt[,foo(.SD), by= "grps"],
parallel= foo2(dt),
times= 10L
)
Unit: seconds
expr min lq mean median uq max neval cld
serial 6.962188 7.312666 8.433159 8.758493 9.287294 9.605387 10 b
parallel 6.563674 6.648749 6.976669 6.937556 7.102689 7.654257 10 a
print(parallel::detectCores()) # 24
Unit: seconds
expr min lq mean median uq max neval cld
serial 9.014247 9.804112 12.17843 13.17508 13.56914 14.13133 10 a
parallel 10.732106 10.957608 11.17652 11.06654 11.30386 12.28353 10 a

Perfilado:

Podemos usar esta respuesta para proporcionar una respuesta más directa al comentario original de @matt dowle en el perfilado.

Como resultado, vemos que la mayor parte del tiempo de cómputo se maneja por base y no por data.table . data.table operaciones de data.table son, como se esperaba, excepcionalmente rápidas. Si bien algunos podrían argumentar que esto es evidencia de que no hay necesidad de paralelismo dentro de la data.table , creo que este flujo de trabajo / conjunto de operaciones no es atípico. Es decir, tengo la firme sospecha de que la mayoría de la agregación de data.table de datos grandes implica una cantidad sustancial de código de data.table de no data.table ; y que esto se correlaciona con el uso interactivo vs uso de desarrollo / producción. Por lo tanto, llego a la conclusión de que el paralelismo sería valioso dentro de los data.table .

library(profr)
prof_list <- replicate(100, profr::profr(dt[,foo(.SD), by= "grps"], interval = 0.002),
simplify = FALSE)
pkg_timing <- fun_timing <- vector("list", length= 100)
for (i in 1:100) {
fun_timing[[i]] <- tapply(prof_list[[i]]$time, paste(prof_list[[i]]$source, prof_list[[i]]$f, sep= "::"), sum)
pkg_timing[[i]] <- tapply(prof_list[[i]]$time, prof_list[[i]]$source, sum)
}
sort(sapply(fun_timing, sum)) # no large outliers
fun_timing2 <- rbindlist(lapply(fun_timing, function(x) {
ret <- data.table(fun= names(x), time= x)
ret[, pct_time := time / sum(time)]
return(ret)
}))
pkg_timing2 <- rbindlist(lapply(pkg_timing, function(x) {
ret <- data.table(pkg= names(x), time= x)
ret[, pct_time := time / sum(time)]
return(ret)
}))
fun_timing2[, .(total_time= sum(time),
avg_time= mean(time),
avg_pct= round(mean(pct_time), 4)), by= "fun"][
order(avg_time, decreasing = TRUE),][1:10,]
pkg_timing2[, .(total_time= sum(time),
avg_time= mean(time),
avg_pct= round(mean(pct_time), 4)), by= "pkg"][
order(avg_time, decreasing = TRUE),]

Resultados:

 fun total_time avg_time avg_pct
1: base::[ 670.362 6.70362 0.2694
2: NA::[.data.table 667.350 6.67350 0.2682
3: .GlobalEnv::foo 335.784 3.35784 0.1349
4: base::[[ 163.044 1.63044 0.0655
5: base::[[.data.frame 133.790 1.33790 0.0537
6: base::%in% 120.512 1.20512 0.0484
7: base::sys.call 86.846 0.86846 0.0348
8: NA::replace_dot_alias 27.824 0.27824 0.0112
9: base::which 23.536 0.23536 0.0095
10: base::sapply 22.080 0.22080 0.0089
pkg total_time avg_time avg_pct
1: base 1397.770 13.97770 0.7938
2: .GlobalEnv 335.784 3.35784 0.1908
3: data.table 27.262 0.27262 0.0155

crossposted en github/data.table

Answer 2

Lo primero que hay que comprobar es que data.table FAQ 3.1 punto 2 se ha hundido en:

Una asignación de memoria se hace solo para el grupo más grande, luego esa memoria se reutiliza para los otros grupos. Hay muy poca basura para recoger.

Esa es una razón por la que la agrupación de tablas de datos es rápida. Pero este enfoque no se presta a la paralelización. Paralelizar significa copiar los datos a los otros subprocesos, en vez de eso, cuesta tiempo. Pero, según tengo entendido, la agrupación data.table suele ser más rápida que plyr con .parallel on de todos modos. Depende del tiempo de cálculo de la tarea para cada grupo, y si ese tiempo de cómputo puede reducirse fácilmente o no. Mover los datos a menudo domina (cuando se realiza una evaluación comparativa de 1 o 3 ejecuciones de tareas de datos grandes).

Más a menudo, hasta ahora, en realidad es algo que está picando en la expresión j de [.data.table . Por ejemplo, recientemente vimos un bajo rendimiento de la agrupación data.table , pero el culpable resultó ser min(POSIXct) ( Agregación en R sobre IDs únicas de 80K ). Evitando ese gotcha rindió más de 50 veces la aceleración.

Entonces el mantra es: Rprof , Rprof , Rprof .

Además, el punto 1 de las mismas preguntas frecuentes podría ser significativo:

Solo esa columna está agrupada, las otras 19 se ignoran porque data.table inspecciona la expresión j y se da cuenta de que no usa las otras columnas.

Entonces, data.table realmente no sigue el paradigma de dividir-aplicar-combinar. Funciona de manera diferente. dividir-aplicar-combinar se presta a la paralelización, pero en realidad no se escala a datos grandes.

También vea la nota 3 en la viñeta de introducción de data.table:

Nos preguntamos cuántas personas están implementando técnicas paralelas para codificar el escaneo vectorial.

Eso es intentar decir "seguro, el paralelo es significativamente más rápido, pero ¿cuánto tiempo debería tomar realmente con un algoritmo eficiente?".

PERO si ha realizado un perfil (utilizando Rprof ), y la tarea por grupo realmente requiere mucho procesamiento, entonces las 3 publicaciones en la ayuda de datos que incluyen la palabra "multinúcleo" pueden ayudar:

publicaciones multinúcleo en datatable-help

Por supuesto, hay muchas tareas donde la paralelización sería agradable en data.table, y hay una manera de hacerlo. Pero aún no se ha hecho, ya que generalmente otros factores influyen, por lo que ha sido una prioridad baja. Si puede publicar datos ficticios reproducibles con puntos de referencia y resultados Rprof, eso ayudaría a aumentar la prioridad.