Objetivo#

El objetivo es realizar el procesamiento de un dataframe aprovechando la capacidad de computo que pueden brindar los nodos del cluster, trabajando todos ellos concurentemente en un proceso necesario dentro de un proceso màs grande (este esquema de trabajo se le conoce como computaciòn paralela), se consigue no solo disponiendo de una arquitectura a nivel fisico y/o a nivel de red que permita conectividad entre los nodos del cluster, sino desde un inicio diseñar aplicaciones pensadas para desempeñar sus actividades de manera descentralizada, este enfoque de desarrollo que reparte los servicios de una aplicaciòn en unos màs pequeños se le llama arquitectura de micro-servicios y es el enfoque que se llevara a cabo en esta aplicaciòn. Por tanto, el primer paso es crear un directorio en el nodo manager al cual tengan acceso todos los nodos trabajadores.

(manager es el termino que se utiliza para referirse al nodo maestro de un cluster creado con Docker Swarm y trabajadores el termino usando para los nodos esclavos, Swarm es el orquestador de contenedores por defecto en Docker, y sobre el cual se construira esta aplicaciòn, asì que se parte del hecho de ya haber creado el Swarm con anterioridad y lo que esto implicaca como; una red que permite el trafico de datos libremente entre los nodos, como una red LAN, W-LAN, la asignaciòn de IPS estaticas, que en el caso de Swarm debe hacerse al menos en el nodo maestro. Para màs informaciòn revisar la documentaciòn oficial de Docker Swarm: https://docs.docker.com/engine/swarm/ )

EL primer paso es crear un volumen compartido al cual puedan acceder todos los nodos, esto se harà por medio de NFS(Network File System), por lo cual se instalara y configurara NFS, esto se harà en el nodo manager(El sistema operativo del manager y los trabajadores es Ubuntu 22.04):

Paso 1: Instalar y configurar NFS#

1
sudo apt update
2
sudo apt install nfs-kernel-server

Paso 2: Construir la imagen y desplegar el servicio#

Crear un directorio para compartir:

1
sudo mkdir -p /srv/nfs/shared
2
sudo chown nobody:nogroup /srv/nfs/shared

Crear un directorio de montaje:

1
sudo mkdir -p /mnt/shared

Montar el NFS: Monta el recurso NFS en el directorio que se acaba de montar (No incluir < > )

1
sudo mount <IP DEL MANAGER>:/srv/nfs/shared /mnt/shared

Verificar la Montura:

1
df -h

Deberìa verse algo asì: :/srv/nfs/shared XXG XXG XXG XX% /mnt/shared La linea “sudo mount :/srv/nfs/shared /mnt/shared” monta manualmente el directorio compartido, pero siempre que se reinicie el dispositivo dejara de estar montado y es necesario ejecutarla de nuevo para que lo este, una forma de automatizar este procedimiento es agregar la configuraciòn en el archivo /etc/fstab. Asì el sistema intentara montarlo automaticamente en cada arranque. Para hacer esto se debe editar el archivo /etc/fstab, de la siguiente manera:

1
sudo nano /etc/fstab

Al final del archivo, agregar una línea para el montaje NFS. Reemplazar :/srv/nfs/shared y /mnt/shared con la ruta correcta del servidor y el punto de montaje en cada sistema si es el caso, por ejemplo:

1
192.168.18.205:/srv/nfs/shared /mnt/shared nfs defaults 0 0

Guardar y salir. Se puede probar que la configuración es correcta montando todos los sistemas de archivos especificados en /etc/fstab con el siguiente comando y sin la necesidad de reiniciar:

1
sudo mount -a

Si no hay errores, el recurso NFS debería aparecer montado en /mnt/shared.

Agregar la configuración a /etc/exports para permitir que los nodos accedan al directorio compartido:

1
echo "/mnt/shared *(rw,sync,no_subtree_check)" | sudo tee -a /etc/exports

Si lo anterior causa problemas, considerando problemas como no poder acceder al directorio compartido desde los otros nodos, ejecutar en el manager:

1
echo "/srv/nfs/shared *(rw,sync,no_subtree_check)" | sudo tee -a /etc/exports

Reiniciar el servicio NFS:

1
sudo exportfs -a
2
sudo systemctl restart nfs-kernel-server

Montar el volumen NFS en los nodos

1
sudo apt install nfs-common

Cear el directorio /mnt/shared en los nodos

1
sudo mkdir -p /mnt/shared

Montar el directorio compartido en cada nodo:

1
sudo mount <IP_DEL_SERVIDOR_NFS>:/srv/nfs/shared /mnt/shared

SCRIPT DE PYTHON QUE SE ENCARGA DE FRAGMENTAR EL DATAFRAME EN TRES PARTES, (POR TANTO MÀS ADELANTE SE DISTRIBUIRA EN TRES NODOS EL PROCESAMIENTO DEL DATAFRAME— EN CASO DE REQUERIR MÀS NODOS SE DEBE MODIFICAR EL SCRIPT EN CONSECUENCIA)

Script para dividir el DataFrame#

Este script divide un archivo CSV en tres partes para procesamiento paralelo.

1
import pandas as pd
2
import numpy as np
3

4
# Cargar el DataFrame completo desde el volumen compartido
5
df = pd.read_csv('/data/Datos_Hidrometeorologicos.csv')
6

7
# Dividir el DataFrame en 3 partes y guardar cada una en el volumen compartido
8
df_split = np.array_split(df, 3)
9
for i, part in enumerate(df_split):
10
    part.to_csv(f'/data/df_part_{i}.csv', index=False)

El script anterior crea 3 dataframes a partir del dataframe Datos_Hidrometeorologicos.csv accede a este a traves del directorio /data del contenedor en el cual se monto el directorio /mnt/shared (esto se define durante la creaciòn del servicio màs adelante). Los dataframes creados tienen el nombre de df_part_0.csv, df_part_1.csv y df_part_2.csv y los almacena en el directorio compartido en /mnt/shared, con el fin de descomponer el dataframe original(Grande) en piezas màs pequeñas que puedan ser procesadas individualente por cada nodo, gracias a que tienen acceso a ellas por medio del directorio compartido, de ahì en principio la importancia de crear el directorio compartido.

A CONTINUCACIÒN SE DEFINE EL DOCKERFILE QUE CONVIERTE EN UNA IMAGEN EL SCRIPT DE PYTHON ANTERIOR Este instala las dependencias necesarias (pandas, numpy) y copia el script split_dataframe.py dentro de la imagen de manera que al iniciar el contenedor se ejecuta el script automaticamente

1
# Dockerfile
2

3
# Usar una imagen base ligera de Python
4
FROM python:3.8-slim
5

6
# Establecer el directorio de trabajo
7
WORKDIR /app
8

9
# Copiar el script Python al contenedor
10
COPY split_dataframe.py .
11

12
# Instalar las dependencias necesarias
13
RUN pip install pandas numpy
14

15
# Ejecutar el script al iniciar el contenedor
16
CMD ["python", "split_dataframe.py"]

Una vez creados los dos archivos anteriores, en el mismo directorio donde se encuentran se construye la imagen de Docker con el siguiente comando:

1
sudo docker build -t split_dataframe_image .

1
sudo docker images

1
sudo docker service create --name fragdata --mount type=bind,source=/mnt/shared,target=/data split_dataframe_image

1
sudo docker save -o /mnt/shared/split_dataframe_image.tar split_dataframe_image:latest

1
sudo docker load -i /mnt/shared/split_dataframe_image.tar

1
sudo docker node update --label-add destino=worker1 worker1

1
sudo docker node ls

1
sudo docker service create --name fragdata --constraint 'node.labels.destino == worker1' --mount type=bind,source=/mnt/shared,target=/data split_dataframe_image

1
sudo docker service update --constraint-rm 'node.labels.destino == worker1' fragdata

1
docker service update --constraint-add 'node.labels.destino == manager' fragdata

1
sudo docker service ps fragdata

Script de procesamiento por fragmento#

1
import pandas as pd
2
import numpy as np
3
import os
4
from collections import Counter
5
import scipy.stats as stats
6

7
# Leer el número de partición del entorno
8
partition = os.getenv('PARTITION', '0')  # Defecto en 0 si no se establece
9

10
# Cargar la partición correspondiente del DataFrame
11
df = pd.read_csv(f'/mnt/shared/df_part_{partition}.csv')
12

13
# Variables para almacenar resultados
14
total_sum = 0
15
count = 0
16
max_value = float('-inf')
17
min_value = float('inf')
18
values_list = []
19
sum_squared_diff = 0
20
rolling_var_sum = 0
21
percentiles = [10, 25, 50, 75, 90]
22

23
# 1. Promedio y varianza
24
for valor in df['ValorObservado']:
25
    total_sum += valor
26
    count += 1
27
temperatura_promedio = total_sum / count
28

29
# 2. Cálculo del máximo y mínimo
30
for valor in df['ValorObservado']:
31
    if valor > max_value:
32
        max_value = valor
33
    if valor < min_value:
34
        min_value = valor
35

36
# 3. Moda
37
for valor in df['ValorObservado']:
38
    values_list.append(valor)
39
frecuencia = Counter(values_list)
40
moda = frecuencia.most_common(1)[0][0]
41

42
# 4. Cálculo de varianza
43
for valor in df['ValorObservado']:
44
    sum_squared_diff += (valor - temperatura_promedio) ** 2
45
varianza = sum_squared_diff / count
46

47
# 5. Desviación estándar
48
desviacion_estandar = np.sqrt(varianza)
49

50
# 6. Calcular percentiles
51
sorted_values = sorted(values_list)
52
percentile_values = {}
53
for p in percentiles:
54
    percentile_values[p] = np.percentile(sorted_values, p)
55

56
# 7. Calcular curtosis y asimetría
57
curtosis = stats.kurtosis(values_list)
58
asimetria = stats.skew(values_list)
59

60
# 8. Ventana móvil para varianza móvil (ventana de 500 elementos)
61
window_size = 500
62
for i in range(window_size, len(values_list)):
63
    window = values_list[i-window_size:i]  # Subconjunto de la ventana
64
    rolling_var_sum += np.var(window)
65

66
# Guardar resultados en un archivo CSV específico para esta partición
67
results = {
68
    "temperatura_promedio": temperatura_promedio,
69
    "max_value": max_value,
70
    "min_value": min_value,
71
    "moda": moda,
72
    "varianza": varianza,
73
    "desviacion_estandar": desviacion_estandar,
74
    "percentiles": percentile_values,
75
    "curtosis": curtosis,
76
    "asimetria": asimetria,
77
    "suma_varianza_movil": rolling_var_sum
78
}
79

80
# Convertir a DataFrame para guardar
81
results_df = pd.DataFrame([results])
82
results_df.to_csv(f'/mnt/shared/result_part_{partition}.csv', index=False)

Este a partir de la variable de entorno PARTITION a la que le define un valor en el despliegue de màs adelante, determinarà que fragmento del dataframe se procesara

DOCKERFILE QUE CONVIERTE EN UNA IMAGEN EL SCRIP ANTERIOR

Dockerfile para procesamiento individual#

1
# Usa una imagen base de Python
2
FROM python:3.8-slim
3

4

5
# Establecer el directorio de trabajo
6
WORKDIR /app
7

8
# Copiar el script de procesamiento al contenedor
9
COPY split_processing.py .
10

11
# Instalar las dependencias necesarias
12
RUN pip install pandas numpy scipy
13

14
# Comando por defecto para ejecutar el script
15
CMD ["python", "split_processing.py"]

En este caso no se desplegará la imagen anterior como un contenedor(run) o un servicio (service create) sino al ser en esta aplicaciòn 3 servicios en el que cada servicio se desplegara en un nodo diferente,por tanto, se usarà docker-compose para desplegar todos al tiempo màs facilmente, ya que organiza la configuraciòn de cada servicio como sus respectivas variables de entorno (PARTITION) y monta el directorio compartido.

docker-compose.yml para procesamiento paralelo#

1
version: '3.8'
2
services:
3
  partition_0:
4
    image: split_processing_image:latest
5
    environment:
6
      - PARTITION=0
7
    deploy:
8
      placement:
9
        constraints: [node.labels.type == manager]
10
      restart_policy:
11
        condition: on-failure
12
    volumes:
13
      - /mnt/shared:/mnt/shared
14

15
  partition_1:
16
    image: split_processing_image:latest
17
    environment:
18
      - PARTITION=1
19
    deploy:
20
      placement:
21
        constraints: [node.labels.type == worker1]
22
      restart_policy:
23
        condition: on-failure
24
    volumes:
25
      - /mnt/shared:/mnt/shared
26

27

28
  partition_2:
29
    image: split_processing_image:latest
30
    environment:
31
      - PARTITION=2
32
    deploy:
33
      placement:
34
        constraints: [node.labels.type == worker2]
35
      restart_policy:
36
        condition: on-failure
37
    volumes:
38
      - /mnt/shared:/mnt/shared

Para desplegar el compose anterior como un stack(compose pero en el marco de docker swarm), se debe tener en un mismo directorio el Dockerfile, split_processing.py y docker-compose.yml.

se debe convertir el script “split_processing.py” en una imagen esto se hace(de debe tener abierto en el terminal el directorio que contiene los tres archivos anteriores, al menos el Dockerfile y el archivo .py):

1
sudo docker build -t split_processing_image .

En este caso a diferencia del primer servicio que fragmenta el dataframe, es obligatorio que la imagen recien creada con el nombre “split_processing_image” este disponible para todos los nodos pues en este caso, sì se desplegara un contenedor de esta imagen en cada uno de ellos, el proceso para hacerlo es el mismo mencionado anteriormene. Guardar en el directorio compartido la imagen en formato .tar:

1
sudo docker save -o /mnt/shared/split_processing_image.tar split_processing_image:latest

Ahora se debe ir a cada nodo y cargar la imagen:

1
sudo docker load -i /mnt/shared/split_processing_image.tar

Despuès en el caso de aùn tener abierto en el terminal el directorio que contiene docker-compose.yml, se puede desplegar el stack:

1
sudo docker stack deploy -c docker-compose.yml stack_processing

Esto desplego un stack con el nombre “stack_processing” conformado por 3 servicios el cual cada uno esta configurado para correr en un nodo diferente. Despuès de un tiempo, el cual depende del tamaño del dataframe, se crearan 3 archivos con el nombre result_part_0.csv, result_part_1.csv y result_part_2.csv que se almacenaran en el directorio compartido, los cuales contienen un dataframe con el resultado del procesamiento individual de cada fragmento. Una vez se tienen estos resultados parciales lo unico que falta es un ultimo servicio que se encargue de reunir estos resultados y operarlos de manera que se conviertan en el resultado total del dataframe original.

Script para combinar resultados#

1
import pandas as pd
2
import ast  # Para convertir el string de percentiles a un diccionario
3

4
# Leer los archivos de resultados de cada partición
5
results = []
6
for i in range(3):
7
    df = pd.read_csv(f'/mnt/shared/result_part_{i}.csv')
8
    results.append(df)
9

10
# Concatenar los DataFrames para facilidad de acceso
11
combined_df = pd.concat(results, ignore_index=True)
12

13
# Calcular el resultado combinado
14
combined_result = {
15
    "temperatura_promedio": combined_df["temperatura_promedio"].mean(),
16
    "max_value": combined_df["max_value"].max(),
17
    "min_value": combined_df["min_value"].min(),
18
    "moda": combined_df["moda"].mode().iloc[0] if not combined_df["moda"].mode().empty else None,
19
    "varianza": combined_df["varianza"].mean(),
20
    "desviacion_estandar": combined_df["desviacion_estandar"].mean(),
21
    "curtosis": combined_df["curtosis"].mean(),
22
    "asimetria": combined_df["asimetria"].mean(),
23
    "suma_varianza_movil": combined_df["suma_varianza_movil"].sum()
24
}
25

26
# Promediar los percentiles
27
percentiles = ["{10: 0, 25: 0, 50: 0, 75: 0, 90: 0}"] * 3  # inicializa en caso de que esté vacío
28
if "percentiles" in combined_df.columns:
29
    percentiles_list = [ast.literal_eval(p) for p in combined_df["percentiles"]]
30
    percentiles_avg = {k: sum(p[k] for p in percentiles_list) / len(percentiles_list) for k in percentiles_list[0]}
31
    combined_result["percentiles"] = percentiles_avg
32

33
# Guardar el resultado combinado en un archivo CSV
34
combined_result_df = pd.DataFrame([combined_result])
35
combined_result_df.to_csv('/mnt/shared/combined_results.csv', index=False)
36

37
print("Resultado combinado guardado en combined_results.csv")

El script anterior crea un archivo llamdo “combined_results.csv” que contiene el resultado final del procesamiento del dataframe

EL SIGUIENTE DOCKERFILE CONVIERTE EN UNA IMAGEN EL SCRIPT ANTERIOR

Dockerfile para combinar resultados#

1
# Usar una imagen de Python
2
FROM python:3.8-slim
3

4

5
# Establecer el directorio de trabajo
6
WORKDIR /app
7

8
# Copiar el script al contenedor
9
COPY collect_results.py .
10

11
# Instalar pandas
12
RUN pip install pandas numpy
13

14
# Comando por defecto
15
CMD ["python", "collect_results.py"]

Despuès se convierte el script en imagen con:

sudo docker build -t result_dataframe_image .

Esto crea una imagen llamada “result_dataframe_image”

Finalmente se crea un servicio que despliega la imagen anterior en el nodo manger

1
sudo docker service create --name dataresult --constraint 'node.labels.destino == manager' --mount type=bind,source=/mnt/shared,target=/mnt/shared result_dataframe_image

El servicio se llama dataresult y casì inmediatamente despuès de ejecutarlo con la linea anterior creara el csv con el resultado, ignorar el hecho de que se este reinicado constantemente y mensajes como “Detected task failure” puesto que el servicio se ejecuta muy rapido, y ya que un contenedor culmina su funciòn se cierra, por tanto esta en constante ciclo de reinicio, asì que se puede cerrar el despliegue con ctr C, y se elimina “combined_results.csv” con(teniendo abierto en el terminal el directorio compartido):

1
sudo rm combined_results.csv

Se cleara uno nuevo casi al instante, mostrando que el servicio se sigue ejecutando correctamente en segundo plano.