Con el incremento en la cantidad de datos y procesamiento en los últimos años, se ha hecho necesario disponer de nuevos paradigmas de computación paralela que permitan repartir la carga de procesamiento en múltiples máquinas mejorando así la eficiencia del proceso.
Fruto de esta necesidad surgió MapReduce y Apache Spark, una extensión de este. Spark es un framework de computación distribuida que planifica, distribuye y monitoriza en varias máquinas (cluster) aplicaciones compuestas de múltiples tareas.
Spark parte de una filosofía modular e integrada ya que dispone de un núcleo básico y varios componentes que interactúan con éste: Spark SQL, MLlib para machine learning, Spark Streaming, gestor de clúster, etc.
En Spark, cada aplicación consiste un programa “driver” que lanza y controla varias operaciones en paralelo en clúster gestionando un número de nodos llamados “executors”. El programa “driver” accede a Spark a través de un objeto contexto, que representa la conexión con el clúster.
Por ejemplo, si necesitáramos realizar un programa que suma varias columnas en un fichero de grandes dimensiones, en Spark cada máquina o nodo realizaría la suma de diferentes partes del fichero.
Una de las principales mejoras de Spark son las RDDs (Resilient distributed dataset), una colección de elementos que se particiona a través de los nodos y que se puede operar en paralelo. Se pueden crear RDDs de dos formas: Paralelizando una colección existente en el programa “driver” (e.g. paralelizando un array o lista creada en el “driver”) o referenciando una base de datos en un sistema de almacenamiento externo soportado por Hadoop como HDFS, HBase, Amazon S3, etc.
Las RDDs soportan dos tipos de operaciones:
- Transformaciones, que crean una nueva RDDs a partir de una existente.
- Acciones, que devuelven un valor al “driver” después de realizar un cálculo en la RDD.
Por ejemplo, map es una transformación que pasa cada elemento de la RDD por una función y devuelve una nueva RDD con el resultado. Reduce es una acción que usa una función para agregar todos los elementos de la RDD devolviendo un resultado. Hay que destacar, que para mejorar la eficiencia, en Spark las transformaciones de las RDDs no se computan hasta que se va a realizar una acción y devolver un resultado al “driver”.
Por otro lado, Spark puede operar en modo local, donde el “driver” y “executor” se ejecutan en la misma máquina, o en modo distribuido en un clúster, usando un gestor de clúster propio o Hadoop YARN, Apache Mesos, etc.
A continuación vamos a ver una pequeña aplicación en modo local de Spark usando la herramienta de Google Colaboratory. Google Colab es un entorno gratuito de Jupyter que se ejecuta en el cloud y permite usar una GPU para mejorar el rendimiento. Es una herramienta ideal para desarrollar y compartir proyectos de machine learning.
Primero de todo es necesario descargar, instalar y lanzar Spark en la máquina virtual siguiendo los pasos mostrados en el siguiente enlace, como se muestra a continuación:
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://www-eu.apache.org/dist/spark/spark-2.3.2/spark-2.3.2-bin-hadoop2.7.tgz
!tar xf spark-2.3.2-bin-hadoop2.7.tgz
!pip install -q findspark
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.3.2-bin-hadoop2.7"
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").getOrCreate()
Ahora es necesario subir el fichero con los datos (son dos columnas en formato Libsvm con dos variables relacionadas linealmente) sobre los que aplicaremos el modelo de regresión lineal de la libreria MLlib de Spark.
from google.colab import files uploaded = files.upload()
Y por último tenemos el código que entrena el modelo de regresión lineal a partir de los datos de entrada.
from pyspark.ml.regression import LinearRegression
# Carga de los datos procedentes del fichero
training = spark.read.format("libsvm")\
.load("sample_linear.txt")
lr = LinearRegression(maxIter=10, regParam=0.01, elasticNetParam=0.01)
# Entrenamiento del modelo
lrModel = lr.fit(training)
# Imprime los coeficientes y el valor constante
print("Coefficients: %s" % str(lrModel.coefficients))
print("Intercept: %s" % str(lrModel.intercept))
# Resumen del modelo y algunas métricas
trainingSummary = lrModel.summary
print("numIterations: %d" % trainingSummary.totalIterations)
print("objectiveHistory: %s" % str(trainingSummary.objectiveHistory))
trainingSummary.residuals.show()
print("RMSE: %f" % trainingSummary.rootMeanSquaredError)
print("r2: %f" % trainingSummary.r2)
Los datos de entrada tenían una relación lineal (con una variable independiente) bastante clara, lo que se ve en los resultados de ejecutar el código, e.g. el pequeño valor de la Raíz del Error Cuadrático Medio (RSME).
Deja una respuesta