Clasificador MultiClase - Apache Spark

Como vimos en el anterior articulo de Blog (Clasificador MultiClase - SciKit Learn) estamos realizando un  experimento de clasificación lo suficientemente complejo para permitirnos probar las distintas tecnologías existentes en el mercado y las diferencias relativas a los siguientes 4 puntos:

1) Performance
2) Consumo de recursos
3) Dificultad para implementar la algorítmica asociada
4) Tiempo de aprendizaje

Para la realización del experimento estamos utilizando un conjunto de datos de caracteres alfanuméricos en formato imagen (28x28) provistos por Google para la realización del curso "Deep Learning by Google" a través de la plataforma Udacity.

Es un conjunto de caracteres NOMINST en distintas formas y formatos simulando lo que sería un reconocimiento de escritura natural sobre 10 clases de caracteres y etiquetados por una clase de 0 a 9. de esta forma la correspondencia será:

A - 0

B - 1

C - 2

 D - 3 

E - 4

F - 5

G - 6

H - 7

I  - 8

J - 9

Para estos experimentos trabajamos con un conjunto de 200.000 datos de entrenamiento y  dos conjuntos de validación y test de 10.000 sucesos. 

Todos los conjuntos han sido previamente mezclados aleatoriamente para el experimento y almacenados en fichero con formato cpickle para reproducir las mismas circunstancias en todos los experimentos.

Apache Spark 2.1.0

Según la definición de la página oficial de Apache Hadoop, 

Spark ™ es un motor de cálculo rápido, distribuido y de uso general para los datos de Hadoop proporcionando un modelo de programación simple y expresivo que soporta una amplia gama de aplicaciones, incluyendo ETL, machine learning, streaming y cálculo de operaciones sobre grafos.

Veamos cada uno de ellos:


Spark SQL: 
componete Spark para procesamiento de datos estructurado. Hay varias maneras de interactuar con Spark SQL, incluyendo SQL y la API Dataset. Cuando se calcula un resultado, se utiliza el mismo motor de ejecución, independientemente de qué API / idioma se utiliza para expresar el cálculo. Esta unificación significa que los desarrolladores pueden cambiar fácilmente de un API a otro, basándose en el que proporciona la forma más natural de expresar una transformación dada.

GraphX: 
componente en Spark para gráficos y cálculo gráfico-paralelo. Para soportar el cálculo de gráficos, GraphX ​​expone un conjunto de operadores fundamentales (por ejemplo, subgraph, joinVertices y aggregateMessages), así como una variante optimizada de la API Pregel. Además, GraphX ​​incluye una creciente colección de algoritmos de gráficos y constructores para simplificar las tareas de análisis gráfico.


MLlib: 
es la biblioteca de aprendizaje automático (ML) de Spark. Su objetivo es hacer que el aprendizaje práctico de la máquina sea escalable y fácil. A un nivel alto, proporciona herramientas tales como:

• ML Algoritmos: algoritmos comunes de aprendizaje tales como clasificación, regresión, agrupación y filtrado colaborativo
• Transformación de Datos: extracción de características, transformación, reducción de dimensionalidad y selección
• Pipelines: herramientas para construir, evaluar y afinar los ductos de ML
• Persistencia: algoritmos de ahorro y carga, modelos y pipelines.
• Otras: Utilidades: álgebra lineal, estadística, manejo de datos, etc.

Spark Streaming: 
es una extensión de la API Spark principal que permite el procesamiento escalable, de alto rendimiento y tolerante a fallos de flujo de datos en vivo. 
Los datos pueden ser ingeridos de muchas fuentes como Kafka, Flume, Kinesis o sockets TCP, y pueden ser procesados ​​usando algoritmos complejos expresados ​​con funciones de alto nivel como mapa, reducir, unir y ventana. 
Los datos procesados ​​pueden ser enviados a sistemas de archivos, bases de datos y cuadros de mando en vivo. 

En nuestro caso utilizaremos el modulo MLlib y el algoritmo LogisticRegression para nuestro experimento.

Inicialización

Lo primero que vamos a hacer es cargar las librerías y módulos necesarios.

# These are all the modules we'll be using later. Make sure you can import them

# before proceeding further.

from __future__ import print_function

import matplotlib.pyplot as plt

import numpy as np

import os

import pickle as pickle

import time

import sys

try:

    # Now we are ready to import Spark Modulestry:

    from pyspark.sql import SparkSession

    from pyspark import SparkContext

    from pyspark import SparkConf

    from pyspark.mllib.classification import LogisticRegressionWithLBFGS as LogisticRegression

    from pyspark.mllib.regression import LabeledPoint

    print("Successfully imported all Spark stuff")

except ImportError as e:

    print("Error importing Spark Modules", e)

    sys.exit(1)

# Config the matplotlib backend as plotting inline in IPython

%matplotlib inline

Successfully imported all Spark stuff

Y Cargamos la imagenes empaquetadas en un fichero cPickle.

Nos basamos en un conjunto de imagenes 28X28 NOMINST (provisto por Google) de distintos tipos de fuentes para los caracteres y etiquetas asociadas:

[A,B,C,D,E,F,G,H,I,J] = [0,1,2,3,4,5,6,7,8,9]

os.chdir('d:/Data/Gdeeplearning-Udacity')

pickle_file = 'notMNIST.pickle'

with open(pickle_file, 'rb') as f:

    save = pickle.load(f)

    train_dataset = save['train_dataset']

    train_labels = save['train_labels']

    valid_dataset = save['valid_dataset']

    valid_labels = save['valid_labels']

    test_dataset = save['test_dataset']

    test_labels = save['test_labels']

    del save  # hint to help gc free up memory

    print('Training set', train_dataset.shape, train_labels.shape)

    print('Validation set', valid_dataset.shape, valid_labels.shape)

    print('Test set', test_dataset.shape, test_labels.shape)

statinfo = os.stat(pickle_file)

print ('Compressed pickle size:', statinfo.st_size)

Como vemos nuestras imágenes y etiquetas han sido perfectamente cargadas y estamos listos para trabajar con ellas.

Training set (200000, 28, 28) (200000,)

Validation set (10000, 28, 28) (10000,)

Test set (10000, 28, 28) (10000,)

Ahora vamos a transformarlas al modelo Spark RDD para que el sistema pueda trabajar con ellos. Aquí viene la primera complicación ya que tenemos que transformar nuestros vectores a un modelo Clave/Valor. Para ello utilizaremos la clase LabeledPoint de forma que como clave estableceremos las etiquetas y como valor el vector de atributos asociados. A priori parece demasiado "artificial" pero es un paso imprescindible para poder utilizar el algoritmo que hemos seleccionado.

def parsePoint(nparray):

   return LabeledPoint(nparray[0], nparray[1:])

conf = SparkConf().setMaster("local[*]").setAppName("LM10").set("spark.executor.memory", "2g") \

    .set("spark.driver.memory", "8g")

    

try:

    spark = SparkContext(conf=conf).getOrCreate()

except Exception as e:

    raise

    spark.stop()

train_dataset = spark.parallelize(np.concatenate((train_labels.reshape(len(train_labels), 1), \                train_dataset.reshape(len(train_dataset), 784)),1)).map(parsePoint)

    

valid_dataset = spark.parallelize(np.concatenate((valid_labels.reshape(len(valid_labels), 1), \                valid_dataset.reshape(len(valid_dataset), 784)),1)).map(parsePoint)

    

sp_test_dataset = 

spark.parallelize(np.concatenate(( test_labels.reshape(len(test_labels)   , 1), \              test_dataset.reshape(len(test_dataset),   784)),1)).map(parsePoint)

Una vez realizado este paso ya estamos preparados para entrenar nuestro modelo. Cabe destacar la configuración necesaria de recursos (set("spark.driver.memory", "8g")) que debemos aplicar para que el sistema no aborte por "out of memory".

Experimentos

Probaremos ahora a entrenar el modelo. Para ello utilizaremos un batch_size de 200 vectores.

batch_size = 200
try:
    # Fit the model
    start = time.time()
    print('Training Dataset : %s' % train_dataset.count())
    num_loops = int(np.ceil(train_dataset.count() / batch_size))
    lrModel = LogisticRegression.train(train_dataset,iterations=num_loops,numClasses=10)
    stop = time.time()
    print('time %s' % (stop - start))

except Exception as e:
    raise
    spark.stop()

Resultados:

Time = 3418.4938788414
Training dataset Accuracy = 0.837155 

Valid dataset Accuracy = 0.8292 

Test dataset Accuracy = 0.8977

Los resultados obtenidos son muy similares a los obtenidos con Scikit-Learn pero el consumo de recursos y el tiempo necesario para generarlos es muy superior (6 superior más para el mejor de los casos scikit-learn).

Validación

Probemos ahora a analizar realmente como esta funcionando el clasificador y la complejidad a la que se enfrenta. Para ello seleccionaremos 9 imágenes al azar y comprobaremos que tal se comporta nuestro modelo.

# Plot the 9 of random results:
labelsAndPreds = sp_test_dataset.map(lambda p: (p.label, lrModel.predict(p.features)))
labelsAndPreds=np.array(labelsAndPreds.collect())
print(labelsAndPreds)
random_index = np.random.randint(0,10000-1,9)
labelsAndPreds = labelsAndPreds [random_index]

Nrows = 3
Ncols = 3
for i in range(9):
    plt.subplot(Nrows, Ncols, i+1)
    plt.imshow(test_dataset[random_index[i]].reshape(28,28), cmap='Greys_r')
    plt.title('Actual: ' + str(labelsAndPreds[i][0]) + ' Pred: ' + str(labelsAndPreds[i][1]),fontsize=10)
    frame = plt.gca()
    frame.axes.get_xaxis().set_visible(False)
    frame.axes.get_yaxis().set_visible(False)
plt.show()

Desde el punto de vista de rendimiento

El algoritmo, desde el punto de vista de recursos presenta comportamientos muy degradado con respecto a las pruebas realizadas con scikit-learn. El consumo de recursos de memoria, en la comparación más favorable, duplica a la prueba anterior, aunque si consigue aprovechar mejor el rendimiento y potencia de la CPU de la máquina.

Con todo esto el tiempo de aprendizaje del modelo casi en un 50%  al peor de los casos anteriores (sckit-learn con solver = newton-cg)












Conclusión

Si bien la tecnología aportada por Spark ™ es muy potente en características, los resultados no son especialmente buenos a nivel de rendimiento con una tasa máxima de aciertos cercana al 90% en el problema que estamos analizando.




Dado su alto consumo de recursos no lo hace indicado para problemas de baja complejidad, si bien, su alta capacidad de escalabilidad junto con un alto repertorio de características indica que puede ser muy valido para sistemas en producción donde la homogeneidad tecnológica y la escalabilidad, como medio de agilizar los procesos de aprendizaje, primen por encima del consumo de recursos.




A nivel curva de aprendizaje y complejidad,  Spark ™ no es especialmente intuitivo. Al ser un plataforma Java sigue sus mismos principios lo cual hace que no sea práctico desde el punto de vista de operaciones con matrices (basando todo el framework en el trabajo con métodos y mappings en lugar de utilizar la sobrecarga de operadores) , si bien, para programadores Java, no debe presentar ninguna complejidad adicional.