chunk.apply

Масштабована обробка даних в R

Simon Urbanek

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

chunk.apply()

Абстрагує процес ітерації
Дозволяє паралельне виконання
iotools є основою hmr, який дає змогу обробляти дані на інфраструктурі Apache Hadoop

mstrsplit() читає порції як матриці

# Скористайтесь chunk.apply, щоб отримувати порції рядків з foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Функція для обробки кожної порції
 function(chunk) {
   # Перетворіть порцію на матрицю
   m <- mstrsplit(chunk, type = "numeric", sep = ",")
   # Поверніть суму по стовпцях
   colSums(m)
 }, 
 # Максимальний розмір порції у байтах
 CH.MAX.SIZE = 1e5)

# Отримайте загальну суму
colSums(chunk_col_sums)

dstrsplit() читає порції як датафрейми

# Скористайтесь chunk.apply, щоб отримувати порції рядків з foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Функція для обробки кожної порції
 function(chunk) {
   # Перетворіть порцію на датафрейм
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   # Поверніть суму по стовпцях
   colSums(d)
 }, 
 # Максимальний розмір порції у байтах
 CH.MAX.SIZE = 1e5)

# Отримайте загальну суму
colSums(chunk_col_sums)

Паралелізація chunk.apply()

# Скористайтесь chunk.apply, щоб отримувати порції рядків з foo.csv
chunk_col_sums <- chunk.apply("foo.csv",

 # Функція для обробки кожної порції
 function(chunk) {

   # Перетворіть порцію на датафрейм
   d <- dstrsplit(chunk, col_types = rep("numeric", 3), sep = ",")
   colSums(d)
 }, 
 # 2 процесори читають і обробляють дані
 CH.PARALLEL = 2)

# Отримайте загальну суму
colSums(chunk_col_sums)

Зауваження щодо паралелізації

Збільшення кількості процесорів не завжди пришвидшує код
На одній машині додаткові процесори зазвичай дають щораз менший виграш

Давайте потренуємось!

Масштабована обробка даних в R