chunk.apply

Traitement de données évolutif en R

Simon Urbanek

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

chunk.apply()

Abstrait le processus de boucle
Permet l'exécution parallèle
iotools est à la base de hmr, qui permet de traiter des données sur l'infrastructure Apache Hadoop

mstrsplit() lit les blocs comme des matrices

# Use chunk.apply to get chunks of rows from foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # A function to process each of the chunk
 function(chunk) {
   # Turn the chunk into a matrix
   m <- mstrsplit(chunk, type = "numeric", sep = ",")
   # Return the column sums
   colSums(m)
 }, 
 # Maximum chunk size in bytes
 CH.MAX.SIZE = 1e5)

# Get the total sum
colSums(chunk_col_sums)

dstrsplit() lit les blocs comme des trames de données

# Use chunk.apply to get chunks of rows from foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # A function to process each of the chunk
 function(chunk) {
   # Turn the chunk into a data frame
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   # Return the column sums
   colSums(d)
 }, 
 # Maximum chunk size in bytes
 CH.MAX.SIZE = 1e5)

# Get the total sum
colSums(chunk_col_sums)

Paralléliser chunk.apply()

# Use chunk.apply to get chunks of rows from foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # A function to process each of the chunk
 function(chunk) {

   # Turn the chunk into a data frame
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   colSums(d)
 }, 
 # 2 processors read and process data
 CH.PARALLEL = 2)

# Get the total sum
colSums(chunk_col_sums)

Note sur la parallélisation

Augmenter le nombre de processeurs n'accélère pas toujours votre code
Les gains diminuent souvent quand on ajoute des processeurs sur une seule machine

Passons à la pratique !

Traitement de données évolutif en R