chunk.apply

Processamento de Dados em Escala no R

Simon Urbanek

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

chunk.apply()

Abstrai o processo de loop
Permite execução paralela
iotools é a base de hmr, que permite processar dados na infraestrutura Apache Hadoop

mstrsplit() lê blocos como matrizes

# Use chunk.apply para obter blocos de linhas de foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Uma função para processar cada bloco
 function(chunk) {
   # Converte o bloco em uma matriz
   m <- mstrsplit(chunk, type = "numeric", sep = ",")
   # Retorna as somas das colunas
   colSums(m)
 }, 
 # Tamanho máximo do bloco em bytes
 CH.MAX.SIZE = 1e5)

# Obter a soma total
colSums(chunk_col_sums)

dstrsplit() lê blocos como data frames

# Use chunk.apply para obter blocos de linhas de foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Uma função para processar cada bloco
 function(chunk) {
   # Converte o bloco em um data frame
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   # Retorna as somas das colunas
   colSums(d)
 }, 
 # Tamanho máximo do bloco em bytes
 CH.MAX.SIZE = 1e5)

# Obter a soma total
colSums(chunk_col_sums)

Paralelizando chunk.apply()

# Use chunk.apply para obter blocos de linhas de foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Uma função para processar cada bloco
 function(chunk) {

   # Converte o bloco em um data frame
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   colSums(d)
 }, 
 # 2 processadores leem e processam os dados
 CH.PARALLEL = 2)

# Obter a soma total
colSums(chunk_col_sums)

Observação sobre paralelização

Aumentar o número de processadores nem sempre acelera seu código
Em uma única máquina, há retornos decrescentes ao adicionar mais processadores

Vamos praticar!

Processamento de Dados em Escala no R