chunk.apply

Elaborazione scalabile dei dati in R

Simon Urbanek

Member of R-Core, Lead Inventive Scientist, AT&T Labs Research

chunk.apply()

Astrae il ciclo
Abilita l'esecuzione parallela
iotools è alla base di hmr, che ti permette di elaborare dati su Apache Hadoop

mstrsplit() legge i blocchi come matrici

# Usa chunk.apply per ottenere blocchi di righe da foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Funzione per elaborare ciascun blocco
 function(chunk) {
   # Trasforma il blocco in una matrice
   m <- mstrsplit(chunk, type = "numeric", sep = ",")
   # Restituisci le somme per colonna
   colSums(m)
 }, 
 # Dimensione massima del blocco in byte
 CH.MAX.SIZE = 1e5)

# Ottieni la somma totale
colSums(chunk_col_sums)

dstrsplit() legge i blocchi come data frame

# Usa chunk.apply per ottenere blocchi di righe da foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Funzione per elaborare ciascun blocco
 function(chunk) {
   # Trasforma il blocco in un data frame
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   # Restituisci le somme per colonna
   colSums(d)
 }, 
 # Dimensione massima del blocco in byte
 CH.MAX.SIZE = 1e5)

# Ottieni la somma totale
colSums(chunk_col_sums)

Parallelizzare chunk.apply()

# Usa chunk.apply per ottenere blocchi di righe da foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Funzione per elaborare ciascun blocco
 function(chunk) {

   # Trasforma il blocco in un data frame
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   colSums(d)
 }, 
 # 2 processori leggono ed elaborano i dati
 CH.PARALLEL = 2)

# Ottieni la somma totale
colSums(chunk_col_sums)

Nota sulla parallelizzazione

Aumentare i processori non sempre velocizza il codice
Su una singola macchina, i benefici aggiungendo processori tendono a calare

Esercitiamoci!

Elaborazione scalabile dei dati in R