library(readr)
library(parallel)
library(dplyr)
#library(data.table)

 
# Ruta relativa del archivo que contiene la base de datos
archivo1 = "Desktop/R ZORA Ejemplos/CLASE 1/train_data.csv"


## leer_trozo: permite cargar una parte de las filas de la base de datos contenida en archivo, donde:
  
# - archivo: ruta completa o parcial en disco de la base de datos.
# - parte: el numero de porcion que se leera.
# - longitud: numero de filas de cada parte.
# 


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


# Por ejemplo, para leer los datos en partes de longitud 100, si queremos leer la parte 5; es decir, 
# los datos que van desde la fila 401 a la fila 500, tomamos: parte = 5, longitud = 100

a = leer_trozo(archivo1,5,100)
a[1:5,1:6]

## cuenta: Contabiliza cuantos elementos no son NAs

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

cuenta(a$X11)


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

# funSuma: Devuelve la suma de los elementos de cada columna

funSuma = function(archivo,parte,longitud,quitar) {
  
  M = leer_trozo(archivo,parte,longitud)
  M = select(M, -quitar)
  s = apply(M,2,sum,na.rm = TRUE)
  return(s)
}


##### Ejemplo  5 #####
k = 5 # total de partes a ser leidas
lon = 100000 # longitud de cada parte

# El total de filas a leer sera k*lon
ff = k*lon
ff

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

### Secuencial ###

cuenta_sec = list(k)
suma_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)
  suma_sec[[j]] = funSuma(archivo1,parte = j,longitud = lon, quitar = nonumer)
  
}
t2 = Sys.time()  

# Tiempo secuencial
tf = t2-t1  
tf 

cuenta_sec2 = do.call(rbind,cuenta_sec)
cuenta_sec2

cuenta_sec3 = apply(cuenta_sec2,2,sum,na.rm = TRUE)
names(cuenta_sec3) = NULL
cuenta_sec3


suma_sec2 = do.call(rbind,suma_sec)
suma_sec3 = apply(suma_sec2,2,sum,na.rm = TRUE)
names(suma_sec3) = NULL

media_sec = suma_sec3/cuenta_sec3
media_sec


### Paralelo ###

ncores <- detectCores(logical = FALSE)
cl <- makeCluster(ncores)
clusterExport(cl,c('leer_trozo','read_csv','select','cuenta'))

v1 = Sys.time()
cuenta_par = clusterApply(cl, x = 1:k, fun = funCuenta, archivo=archivo1,longitud = lon, quitar = nonumer)
suma_par = clusterApply(cl, x = 1:k, fun = funSuma, archivo=archivo1,longitud = lon, quitar = nonumer)
v2 = Sys.time()

stopCluster(cl)

# Tiempo paralelo
vf = v2-v1
vf

cuenta_par2 = do.call(rbind,cuenta_par)
cuenta_par3 = apply(cuenta_par2,2,sum,na.rm = TRUE)
names(cuenta_par3) = NULL

suma_par2 = do.call(rbind,suma_par)
suma_par3 = apply(suma_par2,2,sum,na.rm = TRUE)
names(suma_par3) = NULL


media_par = suma_par3/cuenta_par3
media_par

##### Comparación #####
c(tf,vf)

as.numeric(tf)/as.numeric(vf)