### 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")
head(datos,10)

###### Funciones #####
###  Función de limpieza de textos
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)
}


##### Bucle #####
max_nodo = 10 # se probará el desempeño con un número de nodos de 1 a max_nodo
nn = nrow(datos) # cantidad de datos
# indices en format ri
vec = 1:nn
indices = as.ff(vec)

benches_s = rep(0,max_nodo) # Secuencial.Guardar los tiempos respecto al número de nodos
benches_p = rep(0,max_nodo) # Paralelo: Guardar los tiempos respecto al número de nodos

for(j in 1:max_nodo) {
  cat('\n no. de nodos = ',j,'\n')

  # Indices repartidos en trozos
  chu = chunk(indices,length = j)
  chu

##### Secuencial #####


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

benches_s[j] = bench_s$time


##### Paralelo #####
# inicialización de los nodos con "ff": sfInit, sfExport, sfClusterEval
sfInit(parallel=TRUE, cpus = j, type="SOCK") # j es el mismo indice que el for

# 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()

benches_p[j] = bench_p$time

}

benches = rbind(benches_s,benches_p)
rownames(benches) = c('Secuencial','Paralelo')
colnames(benches) = paste(1:max_nodo,'nodos',sep=' ')
benches



###### Gráfico #####

maxi = max(as.numeric(benches)) *1.5

plot(1:max_nodo,benches[1,],lwd=2,col='tomato1',xlab='No. de Nodos',ylab='Tiempo',
     main="Desempeño Secuencial vs Paralelo", ylim=c(0,maxi) ,type='l')
points(1:max_nodo,benches[1,],pch=16,col='tomato4',cex=1.6)
lines(1:max_nodo,benches[2,],lwd=2,col='dodgerblue2')
points(1:max_nodo,benches[2,],pch=16,col='dodgerblue4',cex=1.6)

legend(2,maxi,pch=c(16,16),col=c('tomato1','dodgerblue2'),cex=0.8,
       legend=c('Secuencial','Paralelo'))

# Aceleración

adif = unlist(benches[1,])/unlist(benches[2,])

plot(1:max_nodo,adif,lwd=2,col='tomato1',xlab='No. de Nodos',ylab='Aceleración',
     main="Gráfico de Aceleración",    type='l')
points(1:max_nodo,adif,pch=16,col='tomato4',cex=1.6)  


# Eficiencia 

efic = (adif/(1:max_nodo))*100
maxi2 = max(max(efic),100)

plot(1:max_nodo,efic,lwd=2,col='tomato1',xlab='No. de Nodos',ylab='Eficiencia',ylim=c(0,maxi2),
     main="Gráfico de Eficiencia", type='l')
points(1:max_nodo,efic,pch=16,col='tomato4',cex=1.6)  

abline(h=50,col="green",lty=3,lwd=3) # linea que marca el 50%
text(1.05,51,'50%',col='seagreen4',pos = 3) # para colocar el texto "50%" cerca de la linea del 50%