En el último artículo, hablé sobre lo que es el subprocesamiento múltiple y di ejemplos de su implementación en el lenguaje R al trabajar con la API doSNOW usando los doParallel doSNOW , doParallel y la construcción foreach .

Este artículo es una continuación, pero se puede considerar como una guía fuera de línea para subprocesos múltiples en R. Se me solicitó que lo escribiera por los comentarios recibidos en la primera parte (aquí un agradecimiento especial a Alexey_mosc , SatCat , Ananiev_Genrih ), en el que me dieron una serie de paquetes que representan un enfoque más moderno para implementaciones de subprocesos múltiples en R, hablaremos de ellas más adelante.

Contenido

• Desafío
• Preparación
• Ejemplo de solución de procesamiento en serie, función sapply y paquete purrr
  
  • Función Sapply
  • Paquete Purrr
• Opciones multiproceso para resolver la tarea de recopilar palabras clave de Yandex.Direct
  
  • Paquete paralelo
  • Paquete futuro
  • Envolturas sobre el futuro
    
    • Paquete de aplicación futura.
    • Paquete Furrr
• Prueba de velocidad
• Conclusión
• Encuesta

Desafío

Como ejemplo, tomamos el problema considerado en una publicación anterior , es decir, en modo multiproceso, recopile una lista de palabras clave de 4 cuentas de publicidad Yandex.Direct.

Para trabajar con la API ryandexdirect , utilizaremos el paquete ryandexdirect . La documentación oficial está en el enlace , pero para la implementación de la tarea descrita, solo necesitamos 2 funciones:

• yadirAuth : autorización en la API Yandex.Direct;
• yadirGetKeyWords : descargue una lista de palabras clave de las cuentas publicitarias.

No es solo que elegí el proceso de descarga de palabras clave, el hecho es que esta es una de las operaciones más largas en la API Yandex.Direct. En segundo lugar, en todas las cuentas el número de palabras clave es diferente, por lo tanto, el tiempo para completar esta operación para cada cuenta será muy diferente, en nuestro caso de 1 a 20 segundos.

Preparación

Inicialmente, debe instalar todos los paquetes discutidos en este artículo, para esto puede usar el siguiente código.

 #    install.packages("ryandexdirect") install.packages("tictoc") install.packages("rbenchmark") install.packages("dplyr") install.packages("purrr") install.packages("future") install.packages("promises") install.packages("furrr") install.packages("future.apply") 

Para que las funciones del paquete estén disponibles para usted, debe conectarlo utilizando el comando de library . Para mayor comodidad, conectaré por separado todos los paquetes necesarios en cada ejemplo de código dado.

Creamos un vector que consiste en inicios de sesión Yandex.Direct, desde el cual luego solicitaremos palabras clave:

 logins <- c("login1", "login2", "login3", "login4") 

Para trabajar con la API Yandex.Direct, primero debe pasar por la autorización de cada cuenta, para esto puede usar el siguiente diseño:

 lapply(logins, function(l) { yadirAuth(Login = l)}) 

Después de ejecutar el código anterior, se abrirá un navegador para autorización bajo cada cuenta. Usted confirma el permiso para que ryandexdirect acceda a sus materiales publicitarios. Serás redirigido a la página donde necesitas copiar el código de verificación. Al ingresarlo en la consola R, complete el proceso de autorización. Esta operación se repite para cada inicio de sesión que especificó al crear los inicios de sesión vectoriales.

Algunos usuarios, durante el proceso de autorización, pueden confundirse por el hecho de una redirección a un recurso de terceros, pero no hay peligro para su cuenta en esto, describí este tema con más detalle en el artículo "Cuán seguro es usar paquetes R para trabajar con API de sistemas publicitarios" .

A continuación, consideraremos varios ejemplos de la implementación de la tarea descrita. Cada uno de los cuales comenzará con un código de ejemplo y su explicación adicional. Creo que esta opción será más conveniente para la percepción.

Ejemplo de solución de procesamiento en serie, función sapply y paquete purrr

En el último artículo , cité una solución usando el bucle for como ejemplo. Dado que consideramos el subprocesamiento múltiple utilizando el paquete foreach , cuya sintaxis se asemeja a los bucles, este ejemplo fue apropiado allí, aunque el uso de bucles no es bienvenido por los usuarios de R.

Los paquetes que consideraremos en este artículo recuerdan más las funciones de la familia de aplicación en sintaxis; por lo tanto, daré un ejemplo de una solución en modo serie usándolos.

Función sapply

Para estimar el tiempo de ejecución de los comandos, en cada uno de los enfoques considerados, utilizaremos el paquete tictoc .

 library(tictoc) library(dplyr) tic() #   kw.sapply <- sapply( logins, #  ,     function(x) #        #     { yadirGetKeyWords(Login = x) %>% mutate(login = x) }, simplify = FALSE #     ) toc() #   #       result.sapply <- do.call("rbind", kw.sapply) 

Tiempo de 39.36 sec elapsed : 39.36 sec elapsed

Al principio, la sintaxis de las funciones de la familia apply no es tan fácil de leer como la sintaxis de los bucles, pero de hecho todo es bastante simple.

sapply(X, FUN)

Donde:

• X - Un objeto cuyos elementos iteraremos y usaremos a su vez en cada iteración, en un bucle for se veía así: for(i in X) ;
• FUN : una función en la que sustituiremos cada elemento del objeto X a su vez, si dibujamos una analogía con for , entonces este es el cuerpo del bucle.

En el Ejemplo de código 4 , el vector de inicio de sesión creado anteriormente se pasa al argumento X. Cada elemento del vector de function(x) { yadirGetKeyWords(Login = x) %>% mutate(login = x) } de sesión se pasa a su vez como el único argumento para la función de function(x) { yadirGetKeyWords(Login = x) %>% mutate(login = x) } anónima function(x) { yadirGetKeyWords(Login = x) %>% mutate(login = x) } que se pasó al argumento FUN .

Es decir sapply ejecutará la función especificada en FUN 4 veces, sustituyendo los inicios de sesión en uno uno por uno, y devolverá el resultado en forma de una lista (objeto de la lista de clase) que consta de 4 elementos. Cada elemento es una tabla con una lista de palabras clave recibidas de la cuenta en cada iteración.

1. yadirGetKeyWords(Login = "login1") %>% mutate(login = "login1")
2. yadirGetKeyWords(Login = "login2") %>% mutate(login = "login2")
3. yadirGetKeyWords(Login = "login3") %>% mutate(login = "login3")
4. yadirGetKeyWords(Login = "login4") %>% mutate(login = "login4")

El objeto obtenido usando sapply tiene la siguiente estructura:

 summary(kw.sapply) 

  Length Class Mode login1 19 data.frame list login2 19 data.frame list login3 19 data.frame list login4 19 data.frame list 

Al final de este ejemplo, el comando result.sapply <- do.call("rbind", kw.sapply) combina los 4 elementos de la lista kw.sapply en un marco result.sapply .

 # A tibble: 6,804 x 1 result.sapply$Id $Keyword $AdGroupId $CampaignId $ServingStatus $State <dbl> <fct> <dbl> <int> <fct> <fct> 1 15164230566 ~ 3597453985 39351725 ELIGIBLE ON 2 15164230567  ~ 3597453985 39351725 ELIGIBLE ON 3 15164230568  ~ 3597453985 39351725 ELIGIBLE ON 4 15164230569 ~ 3597453985 39351725 ELIGIBLE ON 5 15164230570 ~ 3597453985 39351725 ELIGIBLE ON 6 15164230571  ~ 3597453985 39351725 ELIGIBLE ON 7 15164230572 ~ 3597453985 39351725 ELIGIBLE ON 8 15164230573  ~ 3597453985 39351725 ELIGIBLE ON 9 15164230574 ~ 3597453985 39351725 ELIGIBLE ON 10 15164230575 ~ 3597453985 39351725 ELIGIBLE ON # ... with 6,794 more rows, and 13 more variables: $Status <fct>, # $StrategyPriority <fct>, $StatisticsSearchImpressions <int>, # $StatisticsSearchClicks <int>, $StatisticsNetworkImpressions <int>, # $StatisticsNetworkClicks <lgl>, $UserParam1 <chr>, $UserParam2 <chr>, # $ProductivityValue <lgl>, $ProductivityReferences <lgl>, $Bid <dbl>, # $ContextBid <dbl>, $login <chr> 

Además de sapply , la familia de funciones *apply incluye: apply , lapply , vapply , mapply y otras.

Paquete purrr

 library(purrr) library(dplyr) library(tictoc) tic() #   result.purrr <- map_df( logins, #  ,     ~ #   function(.x) { yadirGetKeyWords(Login = .x) %>% mutate(login = .x) } ) toc() #   

Tiempo de 35.46 sec elapsed : 35.46 sec elapsed

El paquete purrr es parte del núcleo de la biblioteca tidyverse , escrito por Headley Wickham.

En términos de significado y sintaxis, las funciones principales del paquete son muy similares a sapply , su principal ventaja es la siguiente:

• Las funciones se dividen en familias map , map2 , pmap , walk etc. Las funciones separadas incluidas en la misma familia devuelven el resultado en diferentes formatos: chr , dbl , int , df , etc.
• Las funciones de la familia map2 sobre elementos (iterar) simultáneamente de dos objetos;
• Las funciones de la familia pmap simultáneamente sobre elementos de cualquier cantidad de objetos. Puede pasar una tabla a la entrada del argumento .l (un análogo del argumento X en sapply) , cada columna de los cuales contendrá los valores por los cuales iterará, y que serán sustituidos a su vez en los argumentos de la misma función pasada en .f (el análogo FUN de sapply) .

¿En qué situación necesitamos iterar sobre elementos de varios objetos? Por ejemplo, trabaja con varias cuentas de agentes, y las cuentas publicitarias de las que desea obtener una lista de palabras clave se encuentran dispersas entre ellas. En este caso, puede crear un vector a partir de los nombres de las cuentas de agentes e iterar sobre él, en paralelo con la forma en que clasifica los inicios de sesión de las cuentas publicitarias.

 library(purrr) #      agencies <- c("agency1", NA, "agency2", "agency1") #      #         result.pmap2 <- map2_df(.x = logins, .y = agencies, ~ { yadirGetKeyWords(Login = .x, AgencyAccount = .y) %>% mutate(login = .x) }) 

Ahora imagine la situación de que cuando inició sesión con diferentes cuentas, guardó el archivo con las credenciales en diferentes carpetas, luego necesita iterar inmediatamente en tres objetos: inicios de sesión de cuentas publicitarias, inicios de sesión de cuentas de agentes, la ruta en la que se almacena el archivo con credenciales. Esto se puede hacer con ayuda. pmap familia pmap .

 library(purrr) #  ,       #      TokenPath <- c("C:\\proj1\\tokens", "C:\\yandex\\token", "C:\\yandex\\token", "C:\\my_yandex_acoount") #   pmap.result <- pmap_df(list(Login = logins, AgencyAccount = agencies, TokenPath = TokenPath), yadirGetKeyWords) 

En consecuencia, el resultado de ejecutar las funciones map_df , map2_df y pmap_df es el marco de fecha, y cuando se usan, no se requiere el último paso del ejemplo con sapply ( do.call("rbind", kw.sapply) ).

El código se ha vuelto más compacto y ejecutado un poco más rápido, pero, sin embargo, los enfoques descritos, sapply y sapply , recopilan palabras clave de cada cuenta de forma secuencial. Por lo tanto, el tiempo total de ejecución de esta operación es igual a la suma de las duraciones de la recopilación de datos de las cuatro cuentas.

Tiempo [total] = Tiempo [inicio de sesión1] + Tiempo [inicio de sesión2] + Tiempo [inicio de sesión3] + Tiempo [inicio de sesión4]

Opciones multiproceso para resolver la tarea de recopilar palabras clave de Yandex.Direct

Entonces, si ya ha leído el primer artículo , entonces sabe que el modo de operación multiproceso tiene varias características:

• Cada hilo comienza en una sesión R separada con un entorno de trabajo limpio.
• Por la misma razón, en un proceso de ejecución separado, los paquetes conectados previamente no se transmiten de manera predeterminada.

La exportación de objetos creados en un entorno de trabajo y la conexión de paquetes en cada enfoque se implementa de manera diferente, luego los consideraremos con más detalle.

Paquete parallel

Este paquete se incluyó por primera vez en el paquete R en la versión 2.14.0 y hasta el día de hoy viene con R en sí.

 library(parallel) library(tictoc) #   cl <- makeCluster(4) #      clusterExport(cl = cl, varlist = "logins") #  ,      #  ,       ryandexdirect clusterEvalQ(cl = cl, { library(ryandexdirect) library(dplyr) } ) tic() #   parallel.kw <- parSapplyLB(cl = cl, #   X = logins, # ,     FUN = function(x) { #      #      X yadirGetKeyWords(Login = x) %>% mutate(login = x) }, simplify = F) #    toc() #     #   stopCluster(cl) #      result.parallel <- dplyr::bind_rows(parallel.kw) 

Tiempo de 16.75 sec elapsed : 16.75 sec elapsed

Intentemos analizar el Código 8 . La función makeCluster crea un grupo de 4 procesos. Podemos exportar objetos de nuestro entorno de trabajo principal al clúster creado usando la función clusterExport , para esto necesitamos usar sus argumentos:

• cl : clúster en el que exportaremos objetos
• varlist : un vector de texto que contiene los nombres de los objetos que se exportarán a cada proceso de clúster.

Una forma de conectar los paquetes correctos en cada nodo del clúster es usar la función clusterEvalQ . En nuestro ejemplo, lo usamos para conectar paquetes, pero puede escribir cualquier código R dentro de clusterEvalQ , y se iniciará al comienzo de cada nodo del clúster. Los argumentos para esta función son bastante obvios, debe especificar el clúster y los comandos que se ejecutarán en él.

parSapplyLB es una versión paralela de la función sapply con equilibrio de carga entre los nodos del clúster, también lo usan, pero debe especificar el clúster con el argumento cl .

También en parallel hay otras versiones paralelas de las funciones de la familia *apply : parLapply , parSapply , parApply , etc.

parSapply difiere de parSapplyLB solo en que no tiene equilibrio de carga en los nodos del clúster.

La función stopCluster se usa para detener el clúster creado.

El último comando, dplyr::bind_rows(parallel.kw) combinamos el objeto parallel.kw obtenido usando parSapplyLB en una tabla.

Para Linux, el parallel tiene funciones separadas: mclapply , mcmapply , mcMap . A menudo, en este sistema operativo, los comandos se ejecutan más rápido y el código se vuelve más compacto.

 library(parallel) library(tictic) library(dplyr) library(ryandexdirect) tic() mclapply.kw <- mclapply(logins, FUN = function(x) { #      #      X yadirGetKeyWords(Login = x) %>% mutate(login = x) }, mc.cores = 4) toc() 

Al usar estas funciones, no es necesario iniciar el clúster con el makeCluster . El número de nodos que especifique utilizando el argumento mc.cores . Tampoco es necesario conectar paquetes y exportar objetos; estas operaciones se realizan automáticamente.

Paquete future

Uno de los enfoques más modernos para la programación asincrónica en R.

Un código que en paralelo resolverá nuestro problema con la ayuda del future es lo suficientemente complicado como para comprenderlo. Por lo tanto, analicemos su trabajo en un ejemplo más simple, solicitaremos una lista de palabras clave de una cuenta.

 library(future) #    plan(multiprocess) #      #    future.kw <- future({yadirGetKeyWords(Login = logins[4])}, packages = "ryandexdirect", globals = "logins") #     resolved(future.kw) #     future.result.1 <- value(future.kw) 

Intentemos descubrir el ejemplo del Código 10 . La función de plan permite establecer y cambiar el modo de ejecución de las expresiones dadas, estas son las principales:

• secuencial : este es el modo de operación R habitual; los comandos se ejecutan secuencialmente en la sesión actual;
• multisesión : modo paralelo, los comandos se ejecutarán en las sesiones en ejecución en segundo plano en la máquina actual, mientras que su sesión de trabajo no se bloqueará;
• cluster : modo paralelo, los comandos se ejecutarán en la máquina actual o remota, de forma similar a cómo se implementa en el paquete parallel .

Todo el paquete future se basa en la ejecución de comandos en procesos en segundo plano sin bloquear la sesión actual. Ejecutar la ejecución de comandos sigue la función del mismo nombre future , así que cuando ejecutamos el comando:

 future({yadirGetKeyWords(Login = logins[4])}, packages = "ryandexdirect", globals = "logins") 

Nuestra sesión actual en R no está bloqueada, y el comando se ejecuta en segundo plano, ejecutando otra sesión de R.

Puede verificar el estado actual del proceso de ejecución de una expresión dada utilizando la función resolved . Finalmente, la función de value se usa para obtener el resultado de future ejecución future . Si ejecuta la función de value antes de su ejecución future en una sesión de ejecución paralela, la sesión de trabajo actual se bloqueará hasta que se complete la expresión de sesión paralela.

El ejemplo de trabajo más avanzado es el uso del future junto con las promises .

 library(future) library(promises) #    plan(multiprocess) #      #    future.kw <- future({suppressMessages( yadirGetKeyWords(Login = logins[4]))}, packages = "ryandexdirect", globals = "logins") %...>% #     future, #      nrow() %...>% paste("words loaded") %...>% print() 

El paquete de promises proporciona un conjunto de operadores de tuberías que complementan perfectamente la funcionalidad future .

En el ejemplo del Código 11 , en segundo plano, comenzamos el proceso de descargar palabras clave de una cuenta publicitaria. Además, el operador de canalización %...>% sin bloquear la sesión de trabajo espera future el future y realiza las operaciones restantes. Como resultado de la ejecución del código, al finalizar el trabajo future , la cantidad de palabras clave de la cuenta especificada se mostrará en la consola:

 [1] "1855 words loaded" 

Al final del artículo, se demostrará un ejemplo más ilustrativo de un montón de future y promises .

De manera predeterminada, el paquete future exporta todo el espacio de trabajo a cada sesión de ejecución paralela, pero usted mismo puede especificar una lista de objetos para exportar utilizando el argumento global .

Para conectar paquetes al future debe pasar un vector que contenga sus nombres al argumento de paquetes .

Ahora, de vuelta a nuestra tarea, el siguiente ejemplo de código en modo paralelo cargará una lista de palabras clave de 4 cuentas:

 library(future) library(tictoc) #   plan("multisession", workers = 4) tic() #   futs <- lapply(logins, #      function(i) #        #   future({ yadirGetKeyWords(Login = i) %>% mutate(login = i) }, packages = c("ryandexdirect", "dplyr"))) completed <- sapply(futs, resolved) #    kw <- lapply(futs, value) #    toc() #    #     result.future <- dplyr::bind_rows(kw) 

Tiempo de 14.83 sec elapsed : 14.83 sec elapsed

Para descargar una lista de palabras clave en modo multiproceso de todas las cuentas publicitarias que figuran en los inicios de sesión de vectores , debe ejecutar un future separado en segundo plano. En el Ejemplo de código 12, implementamos esto usando la función lapply .

El resultado de trabajar lapply es una lista de future lanzados. Puede verificar el estado de cada uno utilizando el sapply(futs, resolved) , que devolverá un vector lógico donde VERDADERO significará que future cumple el future y FALSO ese future está actualmente en progreso.

Para obtener resultados de cada future , después de completar su trabajo, utilizamos el lapply(futs, value) .

: result.future <- dplyr::bind_rows(kw) .

future

, (
future ), .

future.apply

future.apply future , .

 library(future.apply) library(tictoc) #    plan("multisession", workers = 4) tic() #   kw.future.apply <- future_sapply(logins, #    ,   function(x) { #     yadirGetKeyWords(Login = x) %>% mutate(login = x) }, simplify = FALSE, #    #   future.packages = c("ryandexdirect", "dplyr"), future.globals = TRUE ) toc() #    

: 17.28 sec elapsed

13 , future.apply future , .

4 : plan("multisession", workers = 4) .

future_sapply logins . Es decir , , sapply , .

future_sapply future.packages . future.globals . , .

furrr

future . purrr , furrr .

 library(furrr) library(tictoc) #   cl <- parallel::makeCluster(4) plan(cluster, workers = cl) tic() #   furrr.kw <- future_map(logins, ~ #   function(.x) yadirGetKeyWords(Login = .x) %>% mutate(login = .x), .options = future_options(packages = c("ryandexdirect", "dplyr"), globals = c())) toc() #    #      result.furrr <-dplyr::bind_rows(furrr.kw) 

: 15.45 sec elapsed

furrr purrr . purrr , .

.options . .options future_options , .

14 packages globals :

 .options = future_options(packages = c("ryandexdirect", "dplyr"), globals = c()) 

rbenchmark .

, future promises . .

, 20 4 () .

= (T[1] * 20) + (T[2] * 20) + (T[N] * 20)

 library(furrr) library(parallel) library(dplyr) library(future) library(ryandexdirect) library(tictoc) library(rbenchmark) #   logins <- c("login1", "login2", "login3", "login4") #        #        par par.furrr <- function(logins) { cl <- parallel::makeCluster(4) plan(cluster, workers = cl) furrr.kw <- future_map(logins, ~ yadirGetKeyWords(Login = .x) %>% mutate(login = .x), .options = future_options(packages = c("ryandexdirect", "dplyr"), globals = c())) result.furrr <-dplyr::bind_rows(furrr.kw) } par.future <- function(logins) { plan("multisession", workers = 4) futs <- lapply(logins, function(i) future({ yadirGetKeyWords(Login = i) %>% mutate(login = i) }, packages = c("ryandexdirect", "dplyr"))) completed <- sapply(futs, resolved) kw <- lapply(futs, value) result.future <- dplyr::bind_rows(kw) } par.future.apply <- function(logins) { plan("multisession", workers = 4) kw.future.apply <- future_sapply(logins, function(x) { yadirGetKeyWords(Login = x) %>% mutate(login = x) }, simplify = FALSE, future.packages = c("ryandexdirect", "dplyr"), future.globals = TRUE ) result.future.apply <- dplyr::bind_rows(kw.future.apply) } par.parallel <- function(logins) { cl <- parallel::makeCluster(4) clusterExport(cl = cl, varlist = "logins") clusterEvalQ(cl = cl, { library(ryandexdirect) library(dplyr) } ) parallel.kw <- parSapplyLB(cl = cl, X = logins, FUN = function(x) { yadirGetKeyWords(Login = x) %>% mutate(login = x) }, simplify = F) stopCluster(cl) result.parallel <- dplyr::bind_rows(parallel.kw) } #          seq seq.apply <- function(logins) { kw.sapply <- sapply( logins, function(x) { yadirGetKeyWords(Login = x) %>% mutate(login = x) }, simplify = FALSE ) result.sapply <- do.call("rbind", kw.sapply) } seq.purrr <- function(logins) { kw.purrr <- map_df( logins, ~ { yadirGetKeyWords(Login = .x) %>% mutate(login = .x) } ) result.purrr <- do.call("rbind", kw.purrr) } #       rbenchmark #   future + promises #  ,       #          plan(list(tweak(multisession, workers = 2), tweak(multisession, workers = 4))) tic() speed.test <- future({ #          within(benchmark(furrr = par.furrr(logins), future = par.future(logins), future.apply = par.future.apply(logins), parallel = par.parallel(logins), apply = seq.apply(logins), purrr = seq.purrr(logins), replications = c(20), columns = c('test', 'replications', 'elapsed'), order = c('elapsed', 'test')), { average = round(elapsed/replications, 2) }) }, packages = c("dplyr", "ryandexdirect", "rbenchmark", "parallel", "purrr", "future", "promises", "furrr", "future.apply"), globals = c("logins", "par.furrr", "par.future", "par.future.apply", "par.parallel", "seq.apply", "seq.purrr")) %...>% print() %...T>% toc() message("My Session is not blocked") 

3370 , .. .

. , future , promises , .

, . "My Session is not blocked", , , .. .

promises :

• %...>% — %>% , . Es decir , resolved , future , value . , print .
• %...T>% — %T>% , , . , , , .. .. print , , .
• %...T!% — .

15 plan tweak ( plan(list(tweak(multisession, workers = 2), tweak(multisession, workers = 4))) ), , 2 , future 4 .

:

 My Session is not blocked test replications elapsed average 4 parallel 20 393.02 19.65 1 furrr 20 402.09 20.10 2 future 20 431.19 21.56 3 future.apply 20 432.29 21.61 5 apply 20 847.77 42.39 6 purrr 20 864.19 43.21 3370.55 sec elapsed 

, parallel , . furrr , future future.apply .

1 , , . , API . .

, 4 , .

Conclusión

R, API.

, API . " R , 1" .

:

• doSNOW / doParallel + foreach
• future + promises
• future.apply / furrr
• parallel

, , .

, R .