ÑAAS

Cómo lo haríamos en R

library(tidyverse)

## ── Attaching packages ─────────────────────────────────── tidyverse 1.2.1 ──

## ✔ ggplot2 3.1.0       ✔ purrr   0.3.2  
## ✔ tibble  2.1.1       ✔ dplyr   0.8.0.1
## ✔ tidyr   0.8.3       ✔ stringr 1.4.0  
## ✔ readr   1.3.1       ✔ forcats 0.4.0

## ── Conflicts ────────────────────────────────────── tidyverse_conflicts() ──
## ✖ dplyr::filter() masks stats::filter()
## ✖ dplyr::lag()    masks stats::lag()

# simulto datos
x <- rnorm(100, 2, 10)
y <- 2 + 3 * x + rnorm(length(x), 1, 3)
df <- data.frame(id = rep(1:100, each = 3),
                 y = y ,
                 x = x)

Hacemos un group_by y un do, ya sé que en R haríamos un lm(y ~ factor(id) * x, data =df ) y punto, pero es por algo lo de hacerlo de otra forma.

coeficientes <- df %>%
  group_by(id) %>%
  do(coef_x = coef(lm(y ~ x, .))[2]) %>%
  unnest(coef_x)

coeficientes

## # A tibble: 100 x 2
##       id coef_x
##    <int>  <dbl>
##  1     1   3.15
##  2     2   2.78
##  3     3   2.85
##  4     4   3.04
##  5     5   3.18
##  6     6   2.45
##  7     7   2.99
##  8     8   7.13
##  9     9   3.75
## 10    10   3.06
## # … with 90 more rows

En Spark

Iniciamos sparklyr y creamos el spark dataframe a partir de df

library(sparklyr)

## 
## Attaching package: 'sparklyr'

## The following object is masked from 'package:purrr':
## 
##     invoke

sc <- spark_connect(master = "local")

df_tbl <- sc %>% sdf_copy_to(df, name = "df_spark", overwrite = TRUE)
df_tbl

## # Source: spark<df_spark> [?? x 3]
##       id      y      x
##    <int>  <dbl>  <dbl>
##  1     1 -18.9   -8.24
##  2     1  27.4    6.79
##  3     1 -28.4  -10.7 
##  4     2  37.4   10.4 
##  5     2  -6.58  -3.45
##  6     2  24.6    9.95
##  7     3  23.4    6.32
##  8     3 -49.5  -17.1 
##  9     3  18.7    8.48
## 10     4  21.1    6.48
## # … with more rows

Y usamos ahora casi la misma sintaxis que antes, pero en vez de usar lm usamos ml_linear_regression de MLlib.

coeficientes_spark <-  df_tbl %>% group_by(id) %>% 
  do(coeficientes = coef(ml_linear_regression(., y ~ x))[[2]])  %>% 
  unnest(coeficientes)

coeficientes_spark

## # A tibble: 100 x 2
##       id coeficientes
##    <int>        <dbl>
##  1    12         2.83
##  2    13         3.20
##  3    14         2.65
##  4    18         3.00
##  5    25         3.08
##  6    37         2.71
##  7    38         3.09
##  8    46         2.76
##  9    50         3.22
## 10    52         2.64
## # … with 90 more rows

Y lo llamo “ÑAAS” porque aunque es verdad que me hace los modelos en spark para cada id el resultado de aplicar coef al modelo me devuelve el coeficiente pero en estructura de datos de R, no en spark y lo que queremos es que los coeficientes estén en otro sparkdataframes. Así que si alguien sabe como hacer esto en scala usando UDF’s o UDAF’s que lo comente por aquí.