Uso de Spark desde R

Big data con Spark y R

El uso de Spark en DataScience es ahora más común por sus grandes beneficios. Algunos son: almacenamiento distribuido, uso de queries como si se estuviera escribiendo en SQL, desarrollo de modelos de machine learning, entre muchos otros.

Rstudio (Posit) ha desarrollado el paquete sparklyr, el cual me parece que es muy completo. Súper recomendado!!

Los ejercicios de este blog provienen del curso de Udemy: “Taming Big Data with Apache Spark and Python - Hands On!” sin embargo se resolvieron con Sparklyr y la conexión se realiza de forma local.

Cargue librerías

library(R6)
library(tidyverse)
library(sparklyr)

Clase R6 spark_conexion.

La clase spark_conexion mantiene toda información y métodos relevantes de la conexión a Spark. Permitiendo reutilizarlas en todo el código.

spark_conexion <- R6::R6Class(
  classname = "conexion",
  public = list(
    initialize = function() {
      return(invisible(self))
    },
    #' @description
    #' Crea nuevo objeto de conexión
    #' @details
    #' Esta función requiere unas variables de ambiente cargadas para poder
    #' functionar.
    connect = function() {
     self$conn_sp <- sparklyr::spark_connect(master = "local")
      return(invisible(self))
    },
    #' @field conn_spark conexion a spack
    conn_sp  = NULL,
    #' @descripcion Método para dplyr::copy_to con conexion a spark
    #' @param tabla_df Data frame
    #' @param tbl_name Nombre de la tabla en spark
    copy_to_sp = function(tabla_df, tbl_name) {
      copy_to(self$conn_sp, tabla_df, tbl_name) 
    },
    #' @descripcion Método para leer tabla de spark
    #' @param tbl_name Nombre de la tabla en spark
    tbl_sp = function(tbl_name) spark_read_table(self$conn_sp, tbl_name)
  )
)

conn <- spark_conexion$new()$connect()

Operaciones básicas con dplyr

Archivo extraído de grouplens.org.

Ventajas de usar spark desde R:

• Se pueden usar los verbos de dplyr
• Lazy evaluation: Ver que en la parte superior del resultado aparece: # Source: spark<?> [?? x 2])

group_by y count

sparklyr::spark_read_text(
  conn$conn_sp, 
  name = "movieLens",
  "../../data/u.data"
) %>% 
  separate(line, c("user id","item id","rating","timestamp"), sep = "\t") %>% 
  group_by(rating) %>% 
  count()

# Source: spark<?> [?? x 2]
  rating     n
  <chr>  <dbl>
1 3      27145
2 1       6110
3 2      11370
4 4      34174
5 5      21201

summarise, arrange

También permite usar summarise, arrange y operaciones dentro de summarise cómo round, mean.

sp_fake_friends <- sparklyr::spark_read_csv(
  conn$conn_sp,
  name = "fakefriends",
  "../../data/fakefriends.csv",
  header = FALSE,
  columns = c("id", "name", "age", "num_friends")
) %>% 
  group_by(age) %>% 
  summarise(num_friends = round(mean(num_friends), 1)) %>% 
  arrange(age)
sp_fake_friends

# Source:     spark<?> [?? x 2]
# Ordered by: age
   age   num_friends
   <chr>       <dbl>
 1 18           343.
 2 19           213.
 3 20           165 
 4 21           351.
 5 22           206.
 6 23           246.
 7 24           234.
 8 25           198.
 9 26           242.
10 27           228.
# … with more rows

filter min

La función dplyr::filter puede entrar en conflicto con la funcion sparklyr::filter

sparklyr::spark_read_csv(
  conn$conn_sp,
  name = "fakefriends",
  "../../data/fakefriends.csv",
  header = FALSE,
  columns = c("id", "name", "age", "num_friends")
) %>% 
  dplyr::filter(age == min(age))

# Source: spark<?> [?? x 4]
  id    name    age   num_friends
  <chr> <chr>   <chr> <chr>      
1 106   Beverly 18    499        
2 115   Dukat   18    397        
3 341   Data    18    326        
4 377   Beverly 18    418        
5 404   Kasidy  18    24         
6 439   Data    18    417        
7 444   Keiko   18    472        
8 494   Kasidy  18    194        

Operaciones con texto

Las operaciones con texto también pueden ser usadas mediante verbos o secuencia tidyverse.

• ft_tokenizer: Esta función permite almacenar las palabras de la fila en una lista.
  
• ft_stop_words_remover: Se eliminan las palabras conexión tales como: a, en, entre, o, aquí, aún, con, de, e, y, hay, ...

sparklyr::spark_read_text(
  conn$conn_sp,
  path = "../../data/Book"
) %>% 
  ft_tokenizer(
    input_col = "line",
    output_col = "word_list"
      
  ) %>% 
  ft_stop_words_remover(
    input_col = "word_list",
    output_col = "wo_stop_words"
  ) %>% 
  mutate(palabra = explode(wo_stop_words)) %>% 
  filter(palabra != "") %>% 
  group_by(palabra) %>% 
  count() %>% 
  filter(palabra != "�") %>% 
  arrange(desc(n)) %>% 
  head(10)

# Source:     spark<?> [?? x 2]
# Ordered by: desc(n)
   palabra      n
   <chr>    <dbl>
 1 business   290
 2 time       168
 3 need       167
 4 new        150
 5 product    128
 6 people     127
 7 get        122
 8 work       120
 9 may        107
10 want       107

Machine Learning

En mi opinión los beneficios que encontré de aplicar ML con sparklyr son:

• Pipelines: Conjunto de pasos que se desean aplicar al modelo en construcción, es decir, las operaciones a la base, la formula del modelo, seleccion el algoritmo a desarrollar (regresión lineal, árbol de decisión).

• Algoritmos: Sparklyr usa la librería de ML de Spark, por ende, cuenta con una gran variedad de algoritmos para ser usados.

• Transformaciones: ft_dplyr_transformer permite aplicar operaciones con dplyr y aplicarlo en el pipeline creado.

Linear Regression

data

Código

sdf_regresion <- sparklyr::spark_read_text(
  conn$conn_sp,
  path = "../../data/regression.txt"
) %>% 
  separate(line, c("x", "y"), ",") %>% 
  mutate(across(where(is.character), as.numeric))

sdf_regresion

# Source: spark<?> [?? x 2]
       x     y
   <dbl> <dbl>
 1 -1.74  1.66
 2  1.24 -1.18
 3  0.29 -0.4 
 4 -0.13  0.09
 5 -0.39  0.38
 6 -1.79  1.73
 7  0.71 -0.77
 8  1.39 -1.48
 9  1.15 -1.43
10  0.13 -0.07
# … with more rows

Modelo

regresion_pipeline <- sparklyr::ml_pipeline(conn$conn_sp) %>%
  sparklyr::ft_r_formula(y ~ x) %>%
  sparklyr::ml_linear_regression()

partitioned_regresion <- sparklyr::sdf_random_split(
  sdf_regresion,
  training = 0.7,
  testing = 0.3
)

fitted_pipeline <- sparklyr::ml_fit(
  regresion_pipeline,
  partitioned_regresion$training
)

predictions <- sparklyr::ml_transform(
  fitted_pipeline,
  partitioned_regresion$testing
)

predictions

# Source: spark<?> [?? x 5]
       x     y features  label prediction
   <dbl> <dbl> <list>    <dbl>      <dbl>
 1 -3.74  3.75 <dbl [1]>  3.75       3.72
 2 -2.89  2.89 <dbl [1]>  2.89       2.88
 3 -2.54  2.39 <dbl [1]>  2.39       2.53
 4 -2.45  2.44 <dbl [1]>  2.44       2.44
 5 -2.29  2.35 <dbl [1]>  2.35       2.28
 6 -2.12  1.9  <dbl [1]>  1.9        2.11
 7 -2.06  1.95 <dbl [1]>  1.95       2.05
 8 -2     2.02 <dbl [1]>  2.02       1.99
 9 -1.97  1.85 <dbl [1]>  1.85       1.96
10 -1.91  1.86 <dbl [1]>  1.86       1.90
# … with more rows