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.

Clasificador MultiClase - SciKit Learn

En los próximos cuatro artículos del blog probaremos a realizar 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 utilizaremos 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

La elección de este tipo de conjunto de datos no es arbitraria o condicionada por ser un problema relativamente típico de Machine Lerarning. Para su tratamiento, las imágenes deben ser linealizadas y tratadas como un vector de 28x28 atributos, esto es, un problema complejo de análisis de instancias de 784 parámetros y con un patrón interno y estructural definido e interrelacionado.

Este problema se asemeja mucho a los problemas industriales a los que nos enfrentarnos y, sin embargo, es mucho más visual y didáctico que estos.

Para estos experimentos trabajaremos 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.

Scikit-learn 0.18.1

Es una librería python orientada a la resolución de problemas complejos asociados al Datamining y al análisis avanzado de datos.

Esta librería ofrece soluciones para la resolución de los siguientes tipos de problemáticas:

1. Clasificación
2. Regresión
3. Agrupación o clustering
4. Reducción de la Dimensión
5. Optimización de Modelos
6. Preprocesado de datos

En nuestro caso utilizaremos el componente LogisticRegression para nuestro experimento y será propagado en las distintas tecnologías.

Inicialización

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

from __future__ import print_function
import matplotlib.pyplot as plt
import numpy as np
import os
import sys
from IPython.display import display, Image
from scipy import ndimage
from six.moves import cPickle as pickle
import time

# Config the matplotlib backend as plotting inline in IPython
%matplotlib inline

Y cargar las imágenes y etiquetas previamente empaquetadas en nuestros vectores de trabajo.

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)

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

Experimentos

Probaremos el algoritmos sobre tres ténicas de optimización (solvers) adecuados, según documentación, a Large Datasets. Veremos así su respuesta tanto en consumo de recursos (tiempo) como en resultados (performance)

Multiclass Logistic regresion sckitlearn con solver = lbfgs

from sklearn.linear_model import LogisticRegression

lr = LogisticRegression(solver='lbfgs',penalty='l2', max_iter=200)

start = time.time()
lr.fit(train_dataset.reshape(len(train_dataset),784),train_labels)
stop = time.time()
print('time %s' % (stop - start))
print('Valid Accuracy: %s'%lr.score(valid_dataset.reshape(len(valid_dataset),784),valid_labels))
print('Train Accuracy: %s'%lr.score(train_dataset.reshape(len(train_dataset),784),train_labels))
print('Test Accuracy: %s'%lr.score(test_dataset.reshape(len(test_dataset),784),test_labels))

Resultados:

time: 511.25009870529175
Valid Accuracy: 0.824
Train Accuracy: 0.831135
Test Accuracy: 0.8936

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.

random_index = np.random.randint(0,len(test_labels)-1,9)
actuals = test_labels[random_index]
predictions = lr.predict(test_dataset[random_index].reshape(len(random_index),784))
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(actuals[i]) + ' Pred: ' + str(predictions[i]),fontsize=10)
    frame = plt.gca()
    frame.axes.get_xaxis().set_visible(False)
    frame.axes.get_yaxis().set_visible(False)
plt.show()

Multiclass Logistic regresion sckitlearn con solver = sag

lr = LogisticRegression(solver='sag',penalty='l2', max_iter=200)

start = time.time()
lr.fit(train_dataset.reshape(len(train_dataset),784),train_labels)
stop = time.time()
print('time %s' % (stop - start))
print('Valid Accuracy: %s'%lr.score(valid_dataset.reshape(len(valid_dataset),784),valid_labels))
print('Train Accuracy: %s'%lr.score(train_dataset.reshape(len(train_dataset),784),train_labels))
print('Test Accuracy: %s'%lr.score(test_dataset.reshape(len(test_dataset),784),test_labels))

Resultados:

time: 942.5608134269714
Valid Accuracy: 0.8244
Train Accuracy: 0.83132
Test Accuracy: 0.8932

Validación:

# Plot the 9 of random results:

actuals = test_labels[random_index]
predictions = lr.predict(test_dataset[random_index].reshape(len(random_index),784))
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(actuals[i]) + ' Pred: ' + str(predictions[i]),fontsize=10)
    frame = plt.gca()
    frame.axes.get_xaxis().set_visible(False)
    frame.axes.get_yaxis().set_visible(False)
plt.show()

Multiclass Logistic regresion sckit-learn con solver = newton-cg

lr = LogisticRegression(solver='newton-cg',penalty='l2', max_iter=200)

start = time.time()
lr.fit(train_dataset.reshape(len(train_dataset),784),train_labels)
stop = time.time()
print('time %s' % (stop - start))
print('Valid Accuracy: %s'%lr.score(valid_dataset.reshape(len(valid_dataset),784),valid_labels))
print('Train Accuracy: %s'%lr.score(train_dataset.reshape(len(train_dataset),784),train_labels))
print('Test Accuracy: %s'%lr.score(test_dataset.reshape(len(test_dataset),784),test_labels))

Resultados:

time: 1919.3264200687408
Valid Accuracy: 0.8244
Train Accuracy: 0.83131
Test Accuracy: 0.8932

Validación:

# Plot the 9 of random results:
actuals = test_labels[random_index]
predictions = lr.predict(test_dataset[random_index].reshape(len(random_index),784))
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(actuals[i]) + ' Pred: ' + str(predictions[i]),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 similares para los solvers lbgfgs y sag no consiguiendo aprovechar la capacidad de CPU total disponible en la máquina y teniendo un consumo de memoria similar, como podemos ver en las siguientes imágenes

Sin embargo para el solver newton-ng podemos observar que el consumo de memoria es muy superior para obtener resultados similares.

Conclusión

Si bien la tecnología aportada por scikit-learn es muy sencilla de aprender, los resultados no son especialmente buenos a nivel de rendimiento con una tasa máxima de aciertos del 89% en el problema que estamos analizando.

Dado su bajo consumo de recursos uso puede ser adecuado para experimentos realizados en una estación de trabajo de uso común pero no parece ser la mejor opción para sistemas en producción donde la escalabilidad, el performance y la capacidad de adaptarse en recursos a nuevas problemáticas son condiciones indispensables.