##### como balancear la carga cuando los tiempos de ejecuciu?n de las ########
#####  tareas difieren significativamente unos de otros #########


library(parallel)
library(bench)
library(readr)
library(dplyr)
library(snow) # ya no se usa pero nos ayudar? a visualizar

### EJEMPLO 6

# Se generan 30 valores con distribuci?n uniforme en [0,0.5]
set.seed(1377)
tasktime <- runif(9, min = 0, max = 0.5) # usamos la distribuci?n uniforme 
tasktime


#### BENCH ####
# Se usa el comando Sys.sleep que supender? la ejecuci?n de R durante intervalos de tiempo 
# de duraci?n dada por los valores en el vector tasktime. Se compara lapply vs clusterApply

cl <- makeCluster(detectCores())

bench::mark(
  lapply(tasktime, Sys.sleep), 
  clusterApply(cl, tasktime, Sys.sleep),
  iterations = 1
)

stopCluster(cl)


#### SNOW ####

#  tiempo para snow 

#secuencial
time_lapply =  snow.time(
  lapply(tasktime, Sys.sleep)
)

time_lapply
plot(time_lapply)

# paralelo
cl <- makeCluster(detectCores())
time_clusterApply <- snow.time(
  clusterApply(cl, tasktime, Sys.sleep)
)

time_clusterApply
plot(time_clusterApply)

# paralelo LB
time_clusterApplyLB = snow.time(
  clusterApplyLB(cl, tasktime, Sys.sleep)
)

time_clusterApplyLB
plot(time_clusterApplyLB)

stopCluster(cl)

#### EJEMPLO 7

k = 8 # total de partes a ser leidas
lon = 100000 # longitud de cada parte
# Posición de las columnas de la base de datos que no son n?mericas, 
# sino que contienen una etiqueta, y no deben incluirse en los c?lculos
nonumer = c(1,2,54,55,61,63,156,158,159,162,168)
# Ruta relativa del archivo que contiene la base de datos
archivo1 = "Desktop/R ZORA Ejemplos/CLASE 1/train_data.csv"

## Funciones
## leer_trozo: permite cargar una parte de las filas de la base de datos contenida en archivo, donde:

leer_trozo = function(archivo,parte,longitud){
  
  trozo = read_csv(archivo, skip = (parte-1)*longitud+1,n_max = longitud,col_names = FALSE)
  return(trozo)
  
}

## cuenta: Contabiliza cuantos elementos no son NAs

cuenta = function(x) sum(!is.na(x))


## funCuenta: Cuantos elementos diferentes de NA hay en cada columna del trozo leido

funCuenta = function(archivo,parte,longitud,quitar) {
  
  M = leer_trozo(archivo,parte,longitud)
  M = select(M, -quitar)
  k = apply(M,2,cuenta)
  return(k)
}
#################### SYS.TIME
cuenta_sec = list(k)
t1 = Sys.time()
for(j in 1:k) {
  cat('\n Parte ',j)
  cuenta_sec[[j]] = funCuenta(archivo1,parte = j,longitud = lon, quitar = nonumer)
  
}

t2 = Sys.time() 

tf = t2-t1  
tf
#################### BENCH


cl <- makeCluster(detectCores())
clusterExport(cl,c('leer_trozo','read_csv','select','cuenta'))

ben1 = bench::mark(
  lapply(1:k, funCuenta, archivo=archivo1,longitud = lon, quitar = nonumer), 
  clusterApply(cl, 1:k, funCuenta, archivo=archivo1,longitud = lon, quitar = nonumer),
  iterations = 1
)  

stopCluster(cl)

ben1
#################### SNOW

#secuencial
time_lapply =  snow.time(
  lapply(1:k, funCuenta, archivo=archivo1,longitud = lon, quitar = nonumer)
)

time_lapply
plot(time_lapply)

# paralelo
cl <- makeCluster(detectCores())
clusterExport(cl,c('leer_trozo','read_csv','select','cuenta'))

time_clusterApply = snow.time(
  clusterApply(cl, 1:k, funCuenta, archivo=archivo1,longitud = lon, quitar = nonumer)
)

time_clusterApply
plot(time_clusterApply)

# paralelo LB
time_clusterApplyLB = snow.time(
  clusterApplyLB(cl, 1:k, funCuenta, archivo=archivo1,longitud = lon, quitar = nonumer)
)

time_clusterApplyLB
plot(time_clusterApplyLB)

stopCluster(cl)