rss rss hh new rss rss hh new

-Поиск по дневнику

-Подписка по e-mail

-Постоянные читатели

BlooDandMilK Elenka2000 Python3 Stock_Market_Trading _ghbhjif balloonic dmitriy1800 ВаШуня Виталий_Хлусов Главный_синоптик

-Статистика

Создан: 17.03.2011
Записей:
Комментариев:
Написано: 51

Отчеты:
Посетители
Поисковые фразы

[Перевод] R и большие данные: использование Replyr

Суббота, 29 Июля 2017 г. 15:37 + в цитатник

replyr — сокращение от REmote PLYing of big data for R (удаленная обработка больших данных в R).

Почему стоит попробовать replyr? Потому что он позволяет применять стандартные рабочие подходы к удаленным данным (базы данных или Spark).

Можно работать так же, как и с локальным data.frame. replyr предоставляет такие возможности:

Обобщение данных: replyr_summary().
Объединение таблиц: replyr_union_all().
Связывание таблиц по строкам: replyr_bind_rows().
Использование функций разделения, объединения, комбинирования (dplyr::do()): replyr_split(), replyr::gapply().
Аггрегирование/распределение: replyr_moveValuesToRows() / replyr_moveValuesToColumns().
Отслеживание промежуточных результатов.
Контроллер объединений.

Скорее всего, вы всё это делаете с данными локально, поэтому такие возможности сделают работу со Spark и sparklyr гораздо легче.

replyr — продукт коллективного опыта использования R в прикладных решениях для многих клиентов, сбора обратной связи и исправления недостатков.

Примеры ниже.

Все сейчас быстро меняется, поэтому воспользуемся для примеров версиями пакетов в разработке.

base::date()

## [1] "Thu Jul  6 15:56:28 2017"

# devtools::install_github('rstudio/sparklyr')
# devtools::install_github('tidyverse/dplyr')
# devtools::install_github('tidyverse/dbplyr')
# install.packages("replyr")
suppressPackageStartupMessages(library("dplyr"))
packageVersion("dplyr")

## [1] '0.7.1.9000'

packageVersion("dbplyr")

## [1] '1.1.0.9000'

library("tidyr")
packageVersion("tidyr")

## [1] '0.6.3'

library("replyr")
packageVersion("replyr")

## [1] '0.4.2'

suppressPackageStartupMessages(library("sparklyr"))
packageVersion("sparklyr")

## [1] '0.5.6.9012'

# больше памяти, чем предполагается в https://github.com/rstudio/sparklyr/issues/783
config <- spark_config()
config[["sparklyr.shell.driver-memory"]] <- "8G"
sc <- sparklyr::spark_connect(version='2.1.0', 
                              hadoop_version = '2.7',
                              master = "local",
                              config = config)

Summary

Стандартные summary() и glance(), которые нельзя выполнить на Spark.

mtcars_spark <- copy_to(sc, mtcars)

# резюме обработки, а не данных
summary(mtcars_spark)

##     Length Class          Mode
## src 1      src_spark      list
## ops 2      op_base_remote list

packageVersion("broom")

## [1] '0.4.2'

broom::glance(mtcars_spark)

## Error: glance doesn't know how to deal with data of class tbl_sparktbl_sqltbl_lazytbl

replyr_summary работает.

replyr_summary(mtcars_spark) %>%
  select(-lexmin, -lexmax, -nunique, -index)

##    column   class nrows nna    min     max       mean          sd
## 1     mpg numeric    32   0 10.400  33.900  20.090625   6.0269481
## 2     cyl numeric    32   0  4.000   8.000   6.187500   1.7859216
## 3    disp numeric    32   0 71.100 472.000 230.721875 123.9386938
## 4      hp numeric    32   0 52.000 335.000 146.687500  68.5628685
## 5    drat numeric    32   0  2.760   4.930   3.596563   0.5346787
## 6      wt numeric    32   0  1.513   5.424   3.217250   0.9784574
## 7    qsec numeric    32   0 14.500  22.900  17.848750   1.7869432
## 8      vs numeric    32   0  0.000   1.000   0.437500   0.5040161
## 9      am numeric    32   0  0.000   1.000   0.406250   0.4989909
## 10   gear numeric    32   0  3.000   5.000   3.687500   0.7378041
## 11   carb numeric    32   0  1.000   8.000   2.812500   1.6152000

Аггрегирование/распределение

tidyr работает в основном с локальными данными.

mtcars2 <- mtcars %>%
  mutate(car = row.names(mtcars)) %>%
  copy_to(sc, ., 'mtcars2')

# ошибки
mtcars2 %>% 
  tidyr::gather('fact', 'value')

## Error in UseMethod("gather_"): no applicable method for 'gather_' applied to an object of class "c('tbl_spark', 'tbl_sql', 'tbl_lazy', 'tbl')"

mtcars2 %>%
  replyr_moveValuesToRows(nameForNewKeyColumn= 'fact', 
                          nameForNewValueColumn= 'value', 
                          columnsToTakeFrom= colnames(mtcars),
                          nameForNewClassColumn= 'class') %>%
  arrange(car, fact)

## # Source:     lazy query [?? x 4]
## # Database:   spark_connection
## # Ordered by: car, fact
##            car  fact  value   class
##                
##  1 AMC Javelin    am   0.00 numeric
##  2 AMC Javelin  carb   2.00 numeric
##  3 AMC Javelin   cyl   8.00 numeric
##  4 AMC Javelin  disp 304.00 numeric
##  5 AMC Javelin  drat   3.15 numeric
##  6 AMC Javelin  gear   3.00 numeric
##  7 AMC Javelin    hp 150.00 numeric
##  8 AMC Javelin   mpg  15.20 numeric
##  9 AMC Javelin  qsec  17.30 numeric
## 10 AMC Javelin    vs   0.00 numeric
## # ... with 342 more rows

Связывание по строкам

dplyr bind_rows, union и union_all сейчас неприменимы в Spark. replyr::replyr_union_all() и replyr::replyr_bind_rows() — работоспособная альтернатива.

bind_rows()

db1 <- copy_to(sc, 
               data.frame(x=1:2, y=c('a','b'), 
                          stringsAsFactors=FALSE),
               name='db1')
db2 <- copy_to(sc, 
               data.frame(y=c('c','d'), x=3:4, 
                          stringsAsFactors=FALSE),
               name='db2')

# Ошибки из-за попытки осуществить операцию над обработчиком, а не данными
bind_rows(list(db1, db2))

## Error in bind_rows_(x, .id): Argument 1 must be a data frame or a named atomic vector, not a tbl_spark/tbl_sql/tbl_lazy/tbl

union_all

# игнорирует названия столбцов и приводит все данные к строкам
union_all(db1, db2)

## # Source:   lazy query [?? x 2]
## # Database: spark_connection
##       x     y
##    
## 1     1     a
## 2     2     b
## 3     3     c
## 4     4     d

union

# игнорирует названия столбцов и приводит все данные к строкам
# скорее всего, также потеряет дублирующиеся строки
union(db1, db2)

## # Source:   lazy query [?? x 2]
## # Database: spark_connection
##       x     y
##    
## 1     4     d
## 2     1     a
## 3     3     c
## 4     2     b

replyr_bind_rows

replyr::replyr_bind_rows может связывать вместе несколько data.frame-ов.

replyr_bind_rows(list(db1, db2))

## # Source:   table [?? x 2]
## # Database: spark_connection
##       x     y
##    
## 1     1     a
## 2     2     b
## 3     3     c
## 4     4     d

dplyr::do

В нашем примере просто возьмем по нескольку строк из каждой группы аггрегированного набора данных. Обратите внимание: поскольку мы не задаем порядок в явном виде с помощью arrange, нельзя всегда ожидать совпадения результатов в разных источниках данных (БД или Spark).

`dplyr::do` на локальных данных

Из help('do', package='dplyr'):

by_cyl <- group_by(mtcars, cyl)
do(by_cyl, head(., 2))

## # A tibble: 6 x 11
## # Groups:   cyl [3]
##     mpg   cyl  disp    hp  drat    wt  qsec    vs    am  gear  carb
##             
## 1  22.8     4 108.0    93  3.85 2.320 18.61     1     1     4     1
## 2  24.4     4 146.7    62  3.69 3.190 20.00     1     0     4     2
## 3  21.0     6 160.0   110  3.90 2.620 16.46     0     1     4     4
## 4  21.0     6 160.0   110  3.90 2.875 17.02     0     1     4     4
## 5  18.7     8 360.0   175  3.15 3.440 17.02     0     0     3     2
## 6  14.3     8 360.0   245  3.21 3.570 15.84     0     0     3     4

`dplyr::do` на `Spark`

by_cyl <- group_by(mtcars_spark, cyl)
do(by_cyl, head(., 2))

## # A tibble: 3 x 2
##     cyl     V2
##    
## 1     6 
## 2     4 
## 3     8

Получаем не совсем то, что можно использовать.

`replyr` разделение/объединение

mtcars_spark %>%
  replyr_split('cyl', 
               partitionMethod = 'extract') %>%
  lapply(function(di) head(di, 2)) %>%
  replyr_bind_rows()

## # Source:   table [?? x 11]
## # Database: spark_connection
##     mpg   cyl  disp    hp  drat    wt  qsec    vs    am  gear  carb
##             
## 1  21.0     6 160.0   110  3.90 2.620 16.46     0     1     4     4
## 2  21.0     6 160.0   110  3.90 2.875 17.02     0     1     4     4
## 3  22.8     4 108.0    93  3.85 2.320 18.61     1     1     4     1
## 4  24.4     4 146.7    62  3.69 3.190 20.00     1     0     4     2
## 5  18.7     8 360.0   175  3.15 3.440 17.02     0     0     3     2
## 6  14.3     8 360.0   245  3.21 3.570 15.84     0     0     3     4

`replyr gapply`

mtcars_spark %>%
  gapply('cyl',
         partitionMethod = 'extract',
         function(di) head(di, 2))

## # Source:   table [?? x 11]
## # Database: spark_connection
##     mpg   cyl  disp    hp  drat    wt  qsec    vs    am  gear  carb
##             
## 1  21.0     6 160.0   110  3.90 2.620 16.46     0     1     4     4
## 2  21.0     6 160.0   110  3.90 2.875 17.02     0     1     4     4
## 3  22.8     4 108.0    93  3.85 2.320 18.61     1     1     4     1
## 4  24.4     4 146.7    62  3.69 3.190 20.00     1     0     4     2
## 5  18.7     8 360.0   175  3.15 3.440 17.02     0     0     3     2
## 6  14.3     8 360.0   245  3.21 3.570 15.84     0     0     3     4

`replyr::replyr_apply_f_mapped`

Что хотелось бы получить: данные с именами, соответствующими коду (т.е. изменить данные, а не код).

По некоторому размышлению этого можно достигнуть, если связать изменение данных с окружением функции, а не с данными. То есть данные изменены до тех пор, пока соответствующая контролирующая функция выполняется. В нашем случае эта функция — replyr::replyr_apply_f_mapped(), и работает она следующим образом.

Предположим, что операция, которую мы хотим использовать — функция понижения порядка, полученная из некоторого неподконтрольного нам источника (например, пакета). Это может быть простая функция (как ниже), но предположим, мы хотим использовать ее без изменений (исключая даже совсем небольшие, как введение wrapr::let()).

# внешняя функция с заданными в явном виде названиями столбцов
DecreaseRankColumnByOne <- function(d) {
  d$RankColumn <- d$RankColumn - 1
  d
}

Чтобы применить эту функцию к d (в котором не такие, как ожидается, названия столбцов!), мы используем replyr::replyr_apply_f_mapped() для создания нового параметризированного адаптера:

# наши данные
d <- data.frame(Sepal_Length = c(5.8,5.7),
                Sepal_Width = c(4.0,4.4),
                Species = 'setosa',
                rank = c(1,2))

# обработчик для ввода параметров
DecreaseRankColumnByOneNamed <- function(d, ColName) {
  replyr::replyr_apply_f_mapped(d, 
                                f = DecreaseRankColumnByOne, 
                                nmap = c(RankColumn = ColName),
                                restrictMapIn = FALSE, 
                                restrictMapOut = FALSE)
}

# использование
dF <- DecreaseRankColumnByOneNamed(d, 'rank')
print(dF)

##   Sepal_Length Sepal_Width Species rank
## 1          5.8         4.0  setosa    0
## 2          5.7         4.4  setosa    1

replyr::replyr_apply_f_mapped() переименовывает столбцы так, как ожидается в DecreaseRankColumnByOne (соответствие задано в nmap), применяет DecreaseRankColumnByOne и возвращает имена к исходным перед тем, как вернуть результат.

Отслеживание промежуточных результатов

Многие задачи в Sparklyr связаны с созданием промежуточных или временных таблиц. Это можно делать с помощью dplyr::copy_to() и dplyr::compute(). Эти способы могут быть ресурсоёмкими.

В replyr есть функции, позволяющие держать процесс под контролем: генераторы временных имен, не изменяющие собственно данные (они также используются внутри самого пакета).

Сама по себе функция довольно проста:

print(replyr::makeTempNameGenerator)

## function (prefix, suffix = NULL) 
## {
##     force(prefix)
##     if ((length(prefix) != 1) || (!is.character(prefix))) {
##         stop("repyr::makeTempNameGenerator prefix must be a string")
##     }
##     if (is.null(suffix)) {
##         alphabet <- c(letters, toupper(letters), as.character(0:9))
##         suffix <- paste(base::sample(alphabet, size = 20, replace = TRUE), 
##             collapse = "")
##     }
##     count <- 0
##     nameList <- list()
##     function(..., peek = FALSE, dumpList = FALSE, remove = NULL) {
##         if (length(list(...)) > 0) {
##             stop("replyr::makeTempNameGenerator tempname generate unexpected argument")
##         }
##         if (peek) {
##             return(names(nameList))
##         }
##         if (dumpList) {
##             v <- names(nameList)
##             nameList <<- list()
##             return(v)
##         }
##         if (!is.null(remove)) {
##             victims <- intersect(remove, names(nameList))
##             nameList[victims] <<- NULL
##             return(victims)
##         }
##         nm <- paste(prefix, suffix, sprintf("%010d", count), 
##             sep = "_")
##         nameList[[nm]] <<- 1
##         count <<- count + 1
##         nm
##     }
## }
## 
##

Например, чтобы объединить несколько таблиц, хорошее решение для некоторых источников данных — вызывать compute после каждого объединения (иначе полученный SQL может стать длинным и трудным для понимания и в поддержке). Код выглядит примерно так:

# создание данных для примера
names <- paste('table', 1:5, sep='_')
tables <- lapply(names, 
                 function(ni) {
                   di <- data.frame(key= 1:3)
                   di[[paste('val',ni,sep='_')]] <- runif(nrow(di))
                   copy_to(sc, di, ni)
                 })

# собственный генератор временных имён
tmpNamGen <- replyr::makeTempNameGenerator('JOINTMP')

# объединение слева таблиц в последовательности
joined <- tables[[1]]
for(i in seq(2,length(tables))) {
  ti <- tables[[i]]
  if(icode>

Аккуратное введение и управление временными данными может сохранить ресурсы (и время, и место) и значительно улучшить результаты. Нам кажется хорошей практикой задавать в явном виде генератор временных имён, передавать его во все преобразования Sparklyr, а затем очищать временные значения все вместе, когда результаты от них больше не зависят.

Заключение

Если вы хотите тщательно контролировать обработку данных в Spark или БД с помощью R, стоит рассмотреть replyr в дополнение к dplyr и sparklyr.

sparklyr::spark_disconnect(sc)
rm(list=ls())
gc()

##           used (Mb) gc trigger (Mb) max used (Mb)
## Ncells  821292 43.9    1442291 77.1  1168576 62.5
## Vcells 1364897 10.5    2552219 19.5  1694265 13.0

Original source: habrahabr.ru (comments, light).

https://habrahabr.ru/post/334398/

<a href="https://www.liveinternet.ru/users/rss_rss_hh_new/post418692858/">[РџРµСЂРµРІРѕРґ] R Рё Р±РѕР»СЊС€РёРµ РґР°РЅРЅС‹Рµ: РёСЃРїРѕР»СЊР·РѕРІР°РЅРёРµ Replyr</a><br/>replyr вЂ” СЃРѕРєСЂР°С‰РµРЅРёРµ РѕС‚ REmote PLYing of big data for R (СѓРґР°Р»РµРЅРЅР°СЏ РѕР±СЂР°Р±РѕС‚РєР° Р±РѕР»СЊС€РёС… РґР°РЅРЅС‹С… РІ R).

РџРѕС‡РµРјСѓ СЃС‚РѕРёС‚ РїРѕРїСЂРѕР±РѕРІР°С‚СЊ replyr? РџРѕС‚РѕРјСѓ С‡С‚Рѕ РѕРЅ РїРѕР·РІРѕР»СЏРµС‚ РїСЂРёРјРµРЅСЏС‚СЊ СЃС‚Р°РЅРґР°СЂС‚РЅС‹Рµ СЂР°Р±РѕС‡РёРµ РїРѕРґС…РѕРґС‹ Рє СѓРґР°Р»РµРЅРЅС‹Рј РґР°РЅРЅС‹Рј (Р±Р°Р·С‹ РґР°РЅРЅС‹С… РёР»Рё Spark).

РњРѕР¶РЅРѕ СЂР°Р±РѕС‚Р°С‚СЊ С‚Р°Рє Р¶Рµ, РєР°Рє Рё СЃ Р»РѕРєР°Р»СЊРЅС‹Рј data.frame. replyr РїСЂРµРґРѕСЃС‚Р°РІР»СЏРµС‚ С‚Р°РєРёРµ РІРѕР·РјРѕР¶РЅРѕСЃС‚Рё:

РћР±РѕР±С‰РµРЅРёРµ РґР°РЅРЅС‹С…: replyr_summary().
РћР±СЉРµРґРёРЅРµРЅРёРµ С‚Р°Р±Р»РёС†: replyr_union_all().
РЎРІСЏР·С‹РІР°РЅРёРµ С‚Р°Р±Р»РёС† РїРѕ СЃС‚СЂРѕРєР°Рј: replyr_bind_rows().
РСЃРїРѕР»СЊР·РѕРІР°РЅРёРµ С„СѓРЅРєС†РёР№ СЂР°Р·РґРµР»РµРЅРёСЏ, РѕР±СЉРµРґРёРЅРµРЅРёСЏ, РєРѕРјР±РёРЅРёСЂРѕРІР°РЅРёСЏ (dplyr::do()): replyr_split(), replyr::gapply().
РђРіРіСЂРµРіРёСЂРѕРІР°РЅРёРµ/СЂР°СЃРїСЂРµРґРµР»РµРЅРёРµ: ... <a href="https://www.liveinternet.ru/users/rss_rss_hh_new/post418692858/">Р§РёС‚Р°С‚СЊ РґР°Р»РµРµ...</a>

Комментировать

« Пред. запись — К дневнику — След. запись »

Страницы: [1] [Новые]

LiveInternetLiveInternet