### LIBRERIAS ####
library(rtweet)
library(dplyr)
library(stopwords)
library(tidytext)
library(stringr)
library(parallel)
library(ff)
library(bench)
library(snowfall)

###### Tweets previamente guardados
datos = readRDS("estadistica.rds") # En este caso usamos estadística

head(datos,5)

###  Función de limpieza de texto
limpieza_texto = function(tuit){
  # Stopwords a ser eliminados, en ingles y español # elimina las preposiciones 
  # u otros que se repiten y no dan información, verbos etc
  # Stopwords tiene ya unas listas.
  lista_stopwords  = c(stopwords("es"),stopwords("en")) 
  
  # Simbolos a ser removidos
  removidos = "&|<|>" # simbolos que la gente usa: "&|<|>"
  
  # Remocion de simbolos, separación de los textos con formato tweet en palabras, eliminacion de stopwords
  limpio = tuit %>% mutate(texto = str_remove_all(full_text, removidos)) %>%
    unnest_tokens(word, texto, token = "tweets") %>% # separa por palabras: brigramas, trigramas, ngramas, oraciones, etc
    filter(!word %in% lista_stopwords, # me quedo con los que no est?n en la lista del stopwords
           !word %in% str_remove_all(lista_stopwords, "'"), # elimina .....
           str_detect(word,"[a-z]"), # detecta solo palabras con letras y se queda con esas
           str_detect(word,"http",negate = TRUE) )
  
  return(limpio)
  
}
###  Función de conteo por letra inicial
func_words_por_letra = function(D){


# Se limpian las datos y se extraen las palabras en la columna word. 
token_twt = limpieza_texto(D)

# Frecuencia de las palabras
freq_word = token_twt %>%   count(word, sort = TRUE) 

# Se mide cuantos caracteres tiene cada palabra
freq_word$largo = str_length(freq_word$word)

# Se consideran solo palabras con más de 3 letras y menos de 25
palabras = freq_word %>% filter(largo>=3,largo<=25)

# Se extrae la inicial de cada palabra
palabras$inicial = str_sub(palabras$word,1,1)


return(palabras)
}

# número de nodos en los que se repartiran los datos
num_nodo = 4 
# cantidad de datos
nn = nrow(datos)
# indices en format ri
vec = 1:nn
indices = as.ff(vec) # Transforma los datos en un vector tipo ff
# Indices repartidos en trozos
chu = chunk(indices,length = num_nodo)
chu

##### Secuencial #####


bench_s = bench::mark(
  crbind( lapply( chu , function(i) func_words_por_letra(datos[indices[i],])  ) ),
  iterations = 1
)

bench_s$time
palabras_s  = bench_s$result[[1]]

# Unificando todas las partes. El maestro lo hace
palabras2 = palabras_s %>% group_by(word,inicial) %>% summarise(num = sum(n)) %>% arrange(desc(num) )
cuenta = palabras2 %>% group_by(inicial) %>% summarise(cuenta = length(word))
subp = palabras2 %>% group_by(inicial) %>% slice_head(n=1)
cuenta1 = inner_join(cuenta,subp,by="inicial")
names(cuenta1) = c("inicial", "total_palabras",  "mas_repetida" , "repeticiones_mr")
final_s = cuenta1[cuenta1$inicial %in% letters,] 
head(final_s,5)

##### Paralelo #####
# inicialización de los nodos con "ff": sfInit, sfExport, sfClusterEval
sfInit(parallel=TRUE, cpus=num_nodo, type="SOCK")

# Carga de paquetes de R en todos los nodos, incluyendo el maestro.
sfLibrary(ff)
sfLibrary(stopwords)
sfLibrary(dplyr)
sfLibrary(tidytext)
sfLibrary(stringr)

# Carga de funciones creadas y datos
sfExport('func_words_por_letra','limpieza_texto','datos','indices') 



bench_p = bench::mark(
  # sfLapply para ejecutar en paralelo
  crbind( sfLapply(chu,  function(i) func_words_por_letra(datos[indices[i],])  )  ),
  iterations = 1
)


sfStop()

bench_p$time
palabras_p  = bench_p$result[[1]]

# El maestro hace la unificación

# Unificando todas las partes
palabras2 = palabras_p %>% group_by(word,inicial) %>% summarise(num = sum(n)) %>% arrange(desc(num) )
cuenta = palabras2 %>% group_by(inicial) %>% summarise(cuenta = length(word))
subp = palabras2 %>% group_by(inicial) %>% slice_head(n=1)
cuenta1 = inner_join(cuenta,subp,by="inicial")
names(cuenta1) = c("inicial", "total_palabras",  "mas_repetida" , "repeticiones_mr")
final_p = cuenta1[cuenta1$inicial %in% letters,] 
head(final_p,5)

##### COMPARACION ######
benchs = c(bench_s$time,bench_p$time)
names(benchs) = c('Secuencial','Paralelo')
benchs