Deep Learning con PySpark

En este artículo mostraré como crear una Red Neuronal con PySpark con el método de “Multilayer Perceptron” con la cual desarrollaremos una predicción.

¿Que es Multilayer perceptron?

(MLP) es una clase de red neuronal artificial de prealimentación. Un MLP consiste en al menos tres capas de nodos. Excepto por los nodos de entrada, cada nodo es una neurona que usa una función de activación no lineal. MLP utiliza una técnica de aprendizaje supervisado llamada “backpropagation” para entrenamiento. Sus capas múltiples y activación no lineal distinguen a MLP de un perceptrón lineal.
Puede distinguir datos que no son separables linealmente.

Considerando que ya tengas experiencia en Scala o Pyspark esto se hará de la manera mas fácil, en caso todavía estas iniciando estaré publicando algunas funciones básicas en otros articulo para que no te pierdas sobre el uso “ml.feature”, “Pipeline”, “”evaluation”.

PREDICCIÓN DE ANIMALES KAAGLE

Dataset

La fuente del set de datos lo vamos a sacar de kaggle lo puedes obtener desde aquí.

Pero para hacerlo más didáctico he separado el set de datos en 2 partes y la se subido en Google Cloud Store para que puedan descargarlo.

1. Original dataset aquí
2. Zoo_train.csv registros (de 1–80) lo puede descargar aquí
3. Zoo_test.csv registro (81–100) lo puede descargar aquí

1. Tenemos que importar las librerías para el trabajo realizado.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.ml import Pipeline
from pyspark.ml import PipelineModel
from pyspark.ml.feature import VectorAssembler,StringIndexer,VectorIndexer,IndexToString
from pyspark.ml.classification import MultilayerPerceptronClassifier,MultilayerPerceptronClassificationModel
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

2. Crearemos una sesion en Spark como trabajo al cual podremos un nombre.

spark = SparkSession \
 .builder \
 .appName(“RN”) \
 .getOrCreate()

3. Según la condición de kaggle tenemos que nombras las clases por tipo de animal , creamos una función UDF para lograr hacerlo rápido.

def string_to_float(x):
 return float(x)
def condition(r):
 if (r == 1):
 label = “Mamal”
 elif(r==2):
 label = “Bird”
 elif(r==3):
 label = “Reptile”
 elif(r==4):
 label = “Fish”
 elif(r==5):
 label = “Amphibian”
 elif(r==6):
 label = “Bug”
 else:
 label = “Invertebrate”
 return label
string_to_float_udf = F.udf(string_to_float, DoubleType())
animal_udf = F.udf(lambda x: condition(x), StringType())

4. Crear funciones para trabajar en Entrenamiento (Train) y pruebas(Test), por lo general, lo trabajo por separado para subir el código a producción, dependiendo si se desea escalar y mejorar la red neuronal en los pesos.

4.1 Función train.

def get_df_columns_train(df):
 df_train = df.select(F.col(“animal_name”).cast(“String”),
 F.col(“hair”).cast(“Double”),
 F.col(“feathers”).cast(“Double”),
 F.col(“eggs”).cast(“Double”),
 F.col(“milk”).cast(“Double”),
 F.col(“airborne”).cast(“double”),
 F.col(“aquatic”).cast(“double”),
 F.col(“predator”).cast(“double”),
 F.col(“toothed”).cast(“double”),
 F.col(“backbone”).cast(“double”),
 F.col(“breathes”).cast(“double”),
 F.col(“venomous”).cast(“double”),
 F.col(“fins”).cast(“double”),
 F.col(“legs”).cast(“double”),
 F.col(“tail”).cast(“double”),
 F.col(“domestic”).cast(“double”),
 F.col(“catsize”).cast(“double”),
 F.col(“class_type”).cast(“double”).alias(“label”))
 df_train = df_train.withColumn(“label”, animal_udf(F.col(“label”)))
 return df_train

4.2 Función test.

def get_df_columns_test(df):
 df_test = df.select(F.col(“animal_name”).cast(“String”),
 F.col(“hair”).cast(“Double”),
 F.col(“feathers”).cast(“Double”),
 F.col(“eggs”).cast(“Double”),
 F.col(“milk”).cast(“Double”),
 F.col(“airborne”).cast(“double”),
 F.col(“aquatic”).cast(“double”),
 F.col(“predator”).cast(“double”),
 F.col(“toothed”).cast(“double”),
 F.col(“backbone”).cast(“double”),
 F.col(“breathes”).cast(“double”),
 F.col(“venomous”).cast(“double”),
 F.col(“fins”).cast(“double”),
 F.col(“legs”).cast(“double”),
 F.col(“tail”).cast(“double”),
 F.col(“domestic”).cast(“double”),
 F.col(“catsize”).cast(“double”))
 return df_test

5. Crearemos un función en la cual se hará el procesado de todo la información, hay que tener en consideración que los datos ya están limpios. Por que se hará mas sencillo, en otra ocasión hablaremos sobre limpieza de datos con métodos Robustos, fuera de lo convencional como algunos trabajan para que vean la diferencia.

· Esta función trabaja separar el label de los datos.
· Crear un Indexlabel por el atributo del label.
· Crear un vector con los datos de 1:-1 todos menos en id y el label.
· Partición de los datos de (0.8, 0.2) -> train, test.
· Crear nuestra red neuronal con las capas [16, 5, 4, 4, 7].
· Invocamos al perceptron.
· Pipeline es un modo de realizar trabajos con una mayor rapidez.
· Guardamos el modelo.
· Realizamos una evaluación del modelo (Accuracy).

def df_train():
 df = spark.read.format(“com.databricks.spark.csv”)\
 .option(“header”, “true”)\
 .load(“zoo_train.csv”) 
 
 df = get_df_columns_train(df)
 
 assembler = VectorAssembler(inputCols=df.columns[1:-1], outputCol=’features’) 
 features = assembler.transform(df)
 labelIndexer = StringIndexer(inputCol=”label”, outputCol=”indexedLabel”).fit(features)
 featureIndexer = VectorIndexer(inputCol=”features”, outputCol=”indexedFeatures”, maxCategories=3).fit(features)
 
 (trainingData, testData) = features.randomSplit([0.8, 0.2])
 
 layers = [16, 5, 4, 4, 7]
 
 trainer = MultilayerPerceptronClassifier(labelCol=”indexedLabel”, featuresCol=”indexedFeatures”,\
 maxIter=100, layers=layers, blockSize=128, seed=1234)
 labelConverter = IndexToString(inputCol=”prediction”, outputCol=”predictedLabel”,
 labels=labelIndexer.labels)
pipeline = Pipeline(stages=[labelIndexer, featureIndexer, trainer, labelConverter])
 model = pipeline.fit(trainingData)
 predictions = model.transform(testData)
 model.write().overwrite().save(“/jonathan/RN”)
 predictions.select(“prediction”, “indexedLabel”, “features”).show(5)
 evaluator = MulticlassClassificationEvaluator(
 labelCol=”indexedLabel”, predictionCol=”prediction”, metricName=”accuracy”)
 accuracy = evaluator.evaluate(predictions)
 print(“Predictions accuracy = %g, Test Error = %g” % (accuracy,(1.0 — accuracy)))
 
 return model

6. Función test lo que realizamos es cargar el modelo entrenado y poder predecir.

def df_test():
 df = spark.read.format(“com.databricks.spark.csv”)\
 .option(“header”, “true”)\
 .load(“gs://ds-data/file_test/zoo_test.csv”) 
 df = get_df_columns_test(df)
assembler = VectorAssembler(inputCols=df.columns[1:], outputCol=’features’)
 pipeline = Pipeline(stages=[assembler])
 pipelineModel = pipeline.fit(df)
 
 dataset = pipelineModel.transform(df)
 
 loadedPipeline = PipelineModel.read().load(“/jonathan/RN”)
 predictions = loadedPipeline.transform(dataset)
return predictions

7. Ahora para finalizar invocaremos a las funciones para ver el resultado.

Se pudo haber realizado con menos código de linea pero como ejemplo tendría que desglosar las funciones de Spark , Así mismo esta brinda muchas herramientas posibles.

my_model = df_train()

df2 = df_test()
df2_select = df2.select(“animal_name”, “predictedLabel”)
df2_select.toPandas()

Espero le hayas gustado esta publicación sobre redes neuronales con Spark , como científico de datos, tienes esta posibilidad que ya esta implementado Spark con MLlib para que puedas hacer todo, en próximo hablaremos de Pytorch, Keras y posiblemente Tensorflow.js como crear modelo bajo estas herramientas sencillas de utilizar.

linkedin:

Jonathan Quiza | LinkedIn

Gracias