Tutorial PySpark: MLlib con Scikit Learn

Comparativa MLlib y Scikit Learn

Apache Spark MLlib y scikit-learn son dos librerías ampliamente utilizadas en el campo del aprendizaje automático, cada una con sus fortalezas y enfoques particulares. Comprender las diferencias y similitudes entre ellas es esencial para seleccionar la herramienta más adecuada según los requerimientos del proyecto.

Alcance y enfoque

scikit-learn es una librería de Python diseñada para el aprendizaje automático en un solo nodo, ideal para datasets que pueden residir completamente en la memoria de una sola máquina. Ofrece una amplia gama de algoritmos y es conocida por su sencillez y facilidad de uso, lo que la hace perfecta para prototipado rápido y proyectos de menor escala.

Por otro lado, MLlib es la biblioteca de aprendizaje automático de Spark, diseñada para procesamiento distribuido en clusters de computadoras. Está optimizada para manejar grandes volúmenes de datos que superan la capacidad de memoria de una sola máquina, aprovechando el poder de computación en paralelo que ofrece Apache Spark.

Estructuras de datos y procesamiento

scikit-learn utiliza estructuras de datos como NumPy arrays y Pandas DataFrames, operando principalmente en memoria local. Esto puede limitar su capacidad para manejar datasets muy grandes. En contraste, MLlib trabaja con RDDs (Resilient Distributed Datasets) y DataFrames de Spark, permitiendo el procesamiento de datos distribuidos y tolerante a fallos.

Conjunto de algoritmos disponibles

Ambas librerías proporcionan una variedad de algoritmos para tareas de clasificación, regresión, clustering y reducción de dimensionalidad. Sin embargo, scikit-learn ofrece una selección más amplia y madura de algoritmos, incluyendo opciones avanzadas y menos comunes. MLlib, aunque robusta, puede tener implementaciones menos detalladas o configurables, enfocándose en algoritmos que son eficientes en entornos distribuidos.

Rendimiento y escalabilidad

En términos de rendimiento, scikit-learn suele ser más rápida para datasets pequeños debido a la ausencia de sobrecarga de comunicación entre nodos. Sin embargo, su rendimiento disminuye significativamente con datasets grandes que no caben en memoria. MLlib, gracias a su arquitectura distribuida, escalará horizontalmente al agregar más recursos al cluster, manejando eficientemente datasets de gran tamaño.

API y facilidad de uso

La API de scikit-learn es conocida por su consistencia y simplicidad, siguiendo un diseño estandarizado con métodos como fit, predict y transform. Esto facilita el rápido desarrollo y experimentación. MLlib también ofrece una API coherente pero puede requerir una mayor comprensión de Spark y su ecosistema, incluyendo conceptos como SparkSession y pipelines.

Integración y ecosistema

MLlib se integra de manera nativa con otras herramientas dentro del ecosistema de Spark, como Spark SQL y Spark Streaming, permitiendo la construcción de flujos de trabajo complejos que combinan procesamiento de datos en tiempo real y análisis predictivo. scikit-learn, al ser una librería independiente, puede integrarse con otras herramientas de Python pero carece de la cohesión interna que ofrece el ecosistema de Spark.

Paralelización y computación distribuida

Aunque scikit-learn soporta cierta paralelización a través de técnicas como joblib o dask, estas opciones son más limitadas y no se comparan con la capacidad de MLlib para ejecutar tareas de aprendizaje automático a gran escala en un cluster completo. MLlib está diseñada desde cero para la computación distribuida, optimizando tanto el almacenamiento como el procesamiento de datos.

Uso en producción

Para aplicaciones en producción que requieren procesamiento en tiempo real o análisis de big data, MLlib es la opción preferida. Su capacidad para integrarse con sistemas distribuidos y su tolerancia a fallos la hacen adecuada para entornos de alta demanda. scikit-learn es más común en entornos de investigación, desarrollo y en aplicaciones donde el volumen de datos es manejable.

Integración de modelos de scikit-learn en PySpark

La integración de modelos de scikit-learn en flujos de trabajo de PySpark permite combinar lo mejor de ambos mundos: el amplio conjunto de algoritmos y la facilidad de uso de scikit-learn con la escalabilidad y el procesamiento distribuido de PySpark. Esto es especialmente útil cuando se desea aplicar modelos ya entrenados o aprovechar funcionalidades específicas de scikit-learn en entornos de procesamiento de big data.

Para integrar un modelo de scikit-learn en PySpark, es fundamental gestionar adecuadamente la conversión y el movimiento de datos entre las distintas estructuras utilizadas por ambas bibliotecas. Un enfoque común es transformar los DataFrames de Spark en formatos compatibles con scikit-learn, como arrays de NumPy o DataFrames de pandas, y viceversa. Sin embargo, es importante tener en cuenta que estas conversiones pueden ser costosas en términos de rendimiento y memoria si no se manejan con cuidado.

Una práctica recomendada es utilizar Pandas UDFs (User Defined Functions), introducidas en Apache Spark 2.3, que permiten aplicar funciones vectorizadas de pandas sobre particiones de datos de Spark de manera eficiente. Esto facilita la aplicación de modelos de scikit-learn directamente sobre datos distribuidos sin necesidad de recopilar los datos en el controlador.

A continuación, se muestra un ejemplo de cómo integrar un modelo de scikit-learn en un flujo de trabajo de PySpark utilizando Pandas UDFs:

import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf, PandasUDFType
from sklearn.linear_model import LogisticRegression
from sklearn.datasets import load_iris

# Inicializar SparkSession
spark = SparkSession.builder.appName("IntegracionScikitLearn").getOrCreate()

# Cargar datos y entrenar modelo en scikit-learn
iris = load_iris()
X = iris['data']
y = iris['target']
model = LogisticRegression(max_iter=200)
model.fit(X, y)

# Convertir datos a DataFrame de Spark
df = spark.createDataFrame(pd.DataFrame(X, columns=iris['feature_names']))

# Definir Pandas UDF para aplicar el modelo
@pandas_udf("double", PandasUDFType.SCALAR)
def predict_udf(*cols):
    X_pd = pd.concat(cols, axis=1)
    return pd.Series(model.predict(X_pd))

# Aplicar el modelo a los datos de Spark
df_with_predictions = df.withColumn("prediction", predict_udf(*df.columns))
df_with_predictions.show()

En este ejemplo, se entrena un modelo de regresión logística usando scikit-learn y luego se aplica a un DataFrame de Spark mediante un Pandas UDF. La función predict_udf recibe columnas del DataFrame de Spark y las concatena en un DataFrame de pandas, que es el formato esperado por el modelo de scikit-learn. De esta manera, las predicciones se realizan de forma distribuida, aprovechando la paralelización de Spark.

Es importante destacar que el modelo de scikit-learn debe ser serializable para que pueda ser transmitido a los trabajadores de Spark. Esto se logra de manera efectiva si el modelo no contiene referencias a objetos no serializables o recursos externos. Además, para modelos más complejos o si se presentan problemas de serialización, se puede utilizar la clase Broadcast de Spark para enviar el modelo a todos los ejecutores:

from pyspark import SparkContext

sc = SparkContext.getOrCreate()
broadcast_model = sc.broadcast(model)

Luego, dentro del UDF, se puede acceder al modelo mediante broadcast_model.value, asegurando que cada nodo trabajador tenga una copia del modelo sin necesidad de retransmitirlo múltiples veces.

Otra consideración clave es el manejo de grandes volúmenes de datos. Al convertir datos de Spark a pandas o NumPy, existe el riesgo de que los datos no quepan en memoria si las particiones son muy grandes. Para mitigar esto, es crucial ajustar el tamaño de las particiones de Spark utilizando la función repartition() o coalesce(), y asegurarse de que cada partición sea lo suficientemente pequeña para caber en la memoria del ejecutor:

df = df.repartition(100)  # Ajustar el número de particiones según los recursos disponibles

Además, es posible integrar etapas de preprocesamiento y postprocesamiento directamente en el flujo de trabajo de PySpark, utilizando las transformaciones y acciones proporcionadas por Spark. Por ejemplo, se pueden utilizar las funciones de manipulación de datos de PySpark para preparar los datos antes de aplicarlos al modelo de scikit-learn, evitando así transferencias innecesarias de datos.

Para casos en los que se requiera entrenar el modelo de scikit-learn dentro de PySpark, se debe tener en cuenta que el entrenamiento se realizaría en el controlador y no de forma distribuida. Sin embargo, una vez entrenado, el modelo puede ser aplicado de manera distribuida utilizando las técnicas previamente mencionadas.

En escenarios donde se dispone de modelos de scikit-learn previamente entrenados y se desea integrarlos en procesos de ETL (Extract, Transform, Load) o pipelines más amplios en Spark, esta metodología permite incorporar inteligencia artificial en el procesamiento de datos a gran escala. Esto es particularmente útil en aplicaciones de procesamiento en tiempo real o en la generación de características para otros modelos más complejos.

Por último, es recomendable gestionar las dependencias y el entorno de ejecución para asegurar la compatibilidad entre scikit-learn y PySpark. Esto incluye configurar los entornos virtuales o utilizar herramientas como Conda o venv para mantener un control sobre las versiones de las bibliotecas y evitar conflictos en los ejecutores.

Integrar modelos de scikit-learn en flujos de trabajo de PySpark amplía las capacidades analíticas y permite aprovechar modelos previamente desarrollados en entornos más tradicionales de Python. Al combinar estas herramientas, se potencia la flexibilidad y la escalabilidad de las soluciones de análisis de datos y aprendizaje automático en entornos empresariales y de investigación.

Entrenamiento de modelos de scikit-learn en PySpark

El entrenamiento de modelos de scikit-learn en PySpark presenta desafíos y oportunidades al combinar el procesamiento distribuido de Spark con los algoritmos eficientes de scikit-learn. Aunque scikit-learn está diseñado para operar en memoria local, es posible entrenar modelos utilizando datos manejados por PySpark mediante estrategias adecuadas.

Una práctica común es extraer una muestra representativa de los datos para que el entrenamiento sea manejable en términos de memoria. Utilizando métodos como sample() o limit(), se puede obtener un subconjunto que se convierta en un DataFrame de pandas:

# Obtener una muestra de los datos
sample_df = spark_df.sample(fraction=0.1, seed=42)

# Convertir a pandas DataFrame
pandas_df = sample_df.toPandas()

# Separar variables independientes y dependiente
X = pandas_df.drop('etiqueta', axis=1)
y = pandas_df['etiqueta']

# Entrenar el modelo de scikit-learn
from sklearn.ensemble import RandomForestClassifier
modelo = RandomForestClassifier()
modelo.fit(X, y)

Este enfoque es útil cuando el dataset completo es demasiado grande para caber en memoria pero se puede obtener una muestra que capture las características esenciales de los datos.

Otra estrategia consiste en agregar o resumir los datos en PySpark antes de transferirlos a scikit-learn. Por ejemplo, se pueden calcular estadísticas como medias o medianas para reducir la dimensión del dataset:

# Calcular estadísticas agregadas
resumen_df = spark_df.groupBy('categoria').agg({'valor': 'mean'}).toPandas()

# Entrenar el modelo con los datos resumidos
X = resumen_df[['categoria']]
y = resumen_df['avg(valor)']
modelo.fit(X, y)

Al trabajar con datos agregados, se reduce significativamente el volumen de información, facilitando el entrenamiento del modelo sin perder tendencias importantes.

Si el tamaño del dataset lo permite, es posible convertir el DataFrame de Spark completo a pandas utilizando toPandas(). Sin embargo, se debe tener precaución con el consumo de memoria:

# Verificar el número de registros antes de convertir
conteo = spark_df.count()
if conteo < 1e6:
    pandas_df = spark_df.toPandas()
    # Continuar con el entrenamiento
else:
    print("El dataset es demasiado grande para convertir a pandas.")

Es importante monitorear el uso de recursos y ajustar el enfoque según las capacidades del sistema en el que se esté trabajando.

Para escenarios donde se requiere entrenar modelos específicos para diferentes segmentos de datos, se puede utilizar la función mapInPandas() de PySpark para aplicar el entrenamiento en particiones individuales:

def entrenar_modelo(pdf):
    X = pdf.drop('etiqueta', axis=1)
    y = pdf['etiqueta']
    modelo = RandomForestClassifier()
    modelo.fit(X, y)
    pdf['prediccion'] = modelo.predict(X)
    return pdf

# Aplicar la función a cada partición
resultado_df = spark_df.mapInPandas(entrenar_modelo, schema=spark_df.schema)

Este método permite entrenar y aplicar modelos dentro del entorno de Spark, aunque se debe considerar que cada partición entrena un modelo independiente.

También es posible combinar el preprocesamiento en PySpark con el entrenamiento en scikit-learn. Al utilizar PySpark para limpiar y transformar los datos, se puede generar un dataset óptimo para el entrenamiento:

# Preprocesamiento en PySpark
datos_transformados = spark_df.filter("columna > 0").select('columna1', 'columna2', 'etiqueta')

# Conversión a pandas
pandas_df = datos_transformados.toPandas()

# Entrenamiento en scikit-learn
from sklearn.preprocessing import StandardScaler

escalador = StandardScaler()
X = escalador.fit_transform(pandas_df.drop('etiqueta', axis=1))
y = pandas_df['etiqueta']

modelo = RandomForestClassifier()
modelo.fit(X, y)

De esta manera, se aprovechan las fortalezas de PySpark en el manejo y transformación de grandes datasets y las capacidades de scikit-learn en análisis estadístico y modelado.

Es fundamental tener en cuenta que scikit-learn no está diseñado para entrenamiento distribuido. Por lo tanto, al trabajar con datasets que exceden la capacidad de memoria, se debe considerar el uso de MLlib, que está optimizado para entrenamiento en entornos distribuidos.

Finalmente, es recomendable evaluar el flujo de trabajo y determinar si la combinación de PySpark y scikit-learn es la más adecuada para el proyecto. En algunos casos, puede ser más eficiente entrenar modelos con MLlib o utilizar otras bibliotecas especializadas en aprendizaje automático a gran escala.

Paralelización de scikit-learn con Spark

La paralelización de scikit-learn con Spark es una estrategia que busca aprovechar las capacidades de procesamiento distribuido de Apache Spark para ejecutar algoritmos de scikit-learn en entornos de big data. Aunque scikit-learn está diseñado para funcionar sobre un solo nodo, es posible integrar sus modelos con Spark para acelerar el procesamiento y manejar conjuntos de datos de gran tamaño.

Una manera efectiva de lograr esta paralelización es mediante el uso de Spark's mapPartitions, que permite aplicar funciones a cada partición de un DataFrame distribuido. De esta forma, se pueden entrenar múltiples instancias de modelos de scikit-learn en paralelo, procesando diferentes particiones de datos simultáneamente.

Por ejemplo, se puede utilizar la siguiente estrategia:

1. Dividir el conjunto de datos en particiones, donde cada partición contiene un subconjunto de los datos totales.
2. Aplicar una función que entrene un modelo de scikit-learn en cada partición, utilizando la función mapPartitions.
3. Recopilar y combinar los modelos entrenados o sus resultados, según sea necesario.

A continuación, se presenta un ejemplo práctico:

from pyspark.sql import SparkSession
from sklearn.linear_model import LogisticRegression
import pickle

# Inicializar SparkSession
spark = SparkSession.builder.appName("ParalelizacionScikitLearn").getOrCreate()

# Supongamos que tenemos un DataFrame de PySpark
df = spark.read.parquet("ruta_a_datos.parquet")

# Función que será aplicada a cada partición
def entrenar_modelo_por_particion(iterator):
    import pandas as pd
    # Convertir el iterador en un DataFrame de pandas
    datos = pd.DataFrame(list(iterator), columns=columnas)

    X = datos.drop('etiqueta', axis=1).values
    y = datos['etiqueta'].values

    # Entrenar modelo de scikit-learn
    modelo = LogisticRegression()
    modelo.fit(X, y)

    # Serializar el modelo
    modelo_serializado = pickle.dumps(modelo)
    yield modelo_serializado

# Especificar las columnas
columnas = df.columns

# Aplicar la función a las particiones del DataFrame
modelos_rdd = df.rdd.mapPartitions(entrenar_modelo_por_particion)

# Recopilar los modelos entrenados
modelos_entrenados = modelos_rdd.collect()

En este ejemplo, se entrena un modelo de regresión logística en cada partición del DataFrame. Cada modelo se serializa utilizando pickle y se recopila al final. Este enfoque permite la paralelización del entrenamiento de los modelos, aprovechando la arquitectura distribuida de Spark.

Sin embargo, es importante tener en cuenta ciertas consideraciones al paralelizar scikit-learn con Spark:

• Limitaciones de memoria: scikit-learn requiere que los datos que procesa quepan en la memoria de una sola máquina. Por lo tanto, cada partición debe ser lo suficientemente pequeña para no exceder la capacidad de memoria del ejecutor.
• Combinación de modelos: en algunos casos, es necesario combinar los modelos entrenados en cada partición en un único modelo global. Esto puede ser complicado, ya que muchos algoritmos de scikit-learn no están diseñados para ser combinados de esta manera. Una solución es entrenar modelos independientes y utilizar técnicas de ensemble para combinar las predicciones.

Otra alternativa para la paralelización es utilizar la librería spark-sklearn, que proporciona herramientas para distribuir tareas de scikit-learn sobre un cluster de Spark. Aunque no es parte de scikit-learn ni de Spark oficialmente, esta librería facilita la integración entre ambas herramientas.

Por ejemplo, para realizar una búsqueda de hiperparámetros paralelizada:

from spark_sklearn import GridSearchCV
from sklearn.datasets import load_iris
from sklearn.svm import SVC

# Inicializar SparkContext
sc = spark.sparkContext

# Cargar datos
iris = load_iris()
X, y = iris.data, iris.target

# Definir el modelo y la búsqueda de hiperparámetros
svc = SVC()
param_grid = {'C': [0.1, 1, 10], 'kernel': ['linear', 'rbf']}

# Realizar GridSearchCV distribuido
grid_search = GridSearchCV(sc, svc, param_grid)
grid_search.fit(X, y)

# Obtener los mejores parámetros
mejores_parametros = grid_search.best_params_

Con spark-sklearn, el proceso de búsqueda en cuadrícula (GridSearchCV) se paraleliza utilizando los recursos del cluster de Spark, lo que acelera significativamente el proceso para conjuntos de datos grandes o para amplios espacios de hiperparámetros.

Es importante destacar que, aunque estas técnicas permiten aprovechar la paralelización horizontal, no transforman los algoritmos de scikit-learn en algoritmos distribuidos en sentido estricto. Los modelos aún se entrenan en porciones de datos que caben en la memoria de un nodo individual.

Otra opción es utilizar joblib con un backend de Spark para paralelizar tareas de scikit-learn:

from joblibspark import register_spark
from sklearn.model_selection import GridSearchCV
from sklearn.ensemble import RandomForestClassifier
from sklearn.utils import parallel_backend

# Registrar Spark como backend de Joblib
register_spark()

# Definir modelo y parámetros
clf = RandomForestClassifier()
param_grid = {'n_estimators': [50, 100, 200], 'max_depth': [5, 10]}

# Realizar GridSearchCV con paralelización
with parallel_backend('spark'):
    grid_search = GridSearchCV(clf, param_grid, cv=5)
    grid_search.fit(X, y)

# Obtener mejores parámetros
mejores_parametros = grid_search.best_params_

En este caso, se utiliza joblibspark para establecer Spark como el backend de joblib, que scikit-learn utiliza para paralelizar operaciones. Esto permite que las tareas de scikit-learn se ejecuten en paralelo en los diferentes nodos del cluster.

Al implementar estas estrategias, se debe considerar:

• Sobrecarga de comunicación: la transferencia de datos entre el controlador y los ejecutores puede introducir latencia y afectar el rendimiento.
• Serialización del modelo: es necesario asegurar que los modelos y funciones sean serializables para ser transmitidos a través del cluster.
• Compatibilidad de versiones: verificar que las versiones de scikit-learn y Spark utilizadas sean compatibles con las librerías intermedias como spark-sklearn o joblibspark.

Al combinar scikit-learn con Spark, se amplía la capacidad de procesamiento y se mejora la eficiencia en el manejo de grandes volúmenes de datos. Sin embargo, también se incrementa la complejidad del flujo de trabajo, por lo que es esencial contar con una comprensión sólida de ambas herramientas y de los principios de computación distribuida.

Finalmente, es recomendable evaluar si la paralelización de scikit-learn con Spark es la solución óptima para el problema a resolver. En ciertos casos, podría ser más apropiado utilizar algoritmos nativos de MLlib, que están diseñados para operar de manera distribuida y pueden ofrecer un mejor rendimiento y escalabilidad en entornos de big data.

Caso práctico: migración de un modelo de scikit-learn a MLlib

En este caso práctico, se abordará el proceso de migración de un modelo de scikit-learn a MLlib de PySpark. El objetivo es adaptar un modelo existente, desarrollado en un entorno local, a un entorno distribuido y escalable, aprovechando las capacidades de Apache Spark para manejar grandes volúmenes de datos.

Contexto del caso práctico

Supongamos que disponemos de un modelo de regresión logística implementado en scikit-learn para predecir la probabilidad de fuga de clientes en una empresa de telecomunicaciones. Este modelo utiliza características como el tiempo de servicio, el tipo de contrato y el historial de pagos. Debido al crecimiento exponencial de los datos y la necesidad de procesamiento en tiempo real, se requiere migrar este modelo a MLlib para mejorar su escalabilidad y rendimiento.

Paso 1: análisis del modelo existente en scikit-learn

El modelo actual se entrena de la siguiente manera:

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression

# Cargar datos
data = pd.read_csv('datos_clientes.csv')

# Seleccionar características y etiqueta
X = data[['tiempo_servicio', 'tipo_contrato', 'historial_pagos']]
y = data['fuga']

# Dividir datos en entrenamiento y prueba
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Entrenar el modelo
modelo_sk = LogisticRegression()
modelo_sk.fit(X_train, y_train)

# Evaluar el modelo
precision = modelo_sk.score(X_test, y_test)
print(f'Precisión del modelo scikit-learn: {precision}')

En este script, se utilizan pandas para la manipulación de datos y scikit-learn para el entrenamiento del modelo. El siguiente paso es replicar este flujo de trabajo en PySpark MLlib.

Paso 2: Configuración del entorno en PySpark

Iniciar una SparkSession y cargar los datos en un DataFrame de Spark:

from pyspark.sql import SparkSession

# Crear SparkSession
spark = SparkSession.builder.appName('MigracionModelo').getOrCreate()

# Cargar datos en un DataFrame de Spark
data_spark = spark.read.csv('datos_clientes.csv', header=True, inferSchema=True)

Ahora, los datos están listos para ser procesados en un entorno distribuido.

Paso 3: Preprocesamiento de datos en PySpark

Convertir las variables categóricas utilizando OneHotEncoder y ensamblar las características:

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler

# Indexar y codificar variables categóricas
indexer = StringIndexer(inputCol='tipo_contrato', outputCol='tipo_contrato_index')
encoder = OneHotEncoder(inputCol='tipo_contrato_index', outputCol='tipo_contrato_vec')

# Aplicar las transformaciones
data_indexed = indexer.fit(data_spark).transform(data_spark)
data_encoded = encoder.transform(data_indexed)

# Ensamblar las características
assembler = VectorAssembler(
    inputCols=['tiempo_servicio', 'tipo_contrato_vec', 'historial_pagos'],
    outputCol='features'
)

data_preparada = assembler.transform(data_encoded)

Se han convertido las variables categóricas y ensamblado todas las características en un vector para su uso en MLlib.

Paso 4: Dividir los datos en entrenamiento y prueba

Dividir el conjunto de datos utilizando randomSplit:

# Dividir los datos
train_data, test_data = data_preparada.randomSplit([0.8, 0.2], seed=42)

Esto asegura que el modelo se entrene y evalúe de manera consistente.

Paso 5: Entrenar el modelo con MLlib

Entrenar el modelo de regresión logística utilizando MLlib:

from pyspark.ml.classification import LogisticRegression

# Crear el modelo
modelo_ml = LogisticRegression(featuresCol='features', labelCol='fuga')

# Entrenar el modelo
modelo_entrenado = modelo_ml.fit(train_data)

El modelo ahora está entrenado utilizando el entorno distribuido de Spark.

Paso 6: Evaluar el modelo migrado

Evaluar el modelo utilizando el conjunto de prueba:

# Realizar predicciones
predicciones = modelo_entrenado.transform(test_data)

# Evaluar el modelo
from pyspark.ml.evaluation import BinaryClassificationEvaluator

evaluador = BinaryClassificationEvaluator(labelCol='fuga', rawPredictionCol='rawPrediction', metricName='areaUnderROC')
auc = evaluador.evaluate(predicciones)
print(f'Área bajo la curva ROC del modelo MLlib: {auc}')

La métrica Área bajo la curva ROC (AUC) es utilizada para comparar el rendimiento con el modelo original.

Paso 7: Comparación de resultados y ajustes necesarios

Comparar los resultados obtenidos con ambos modelos:

• Precisión del modelo scikit-learn: valor obtenido previamente.
• AUC del modelo MLlib: valor calculado en el paso anterior.

Si se observan diferencias significativas, puede ser necesario ajustar los parámetros del modelo o revisar el preprocesamiento de datos para garantizar que ambos modelos sean comparables.

Paso 8: Implementar pipelines en MLlib

Crear un Pipeline en MLlib para automatizar el proceso:

from pyspark.ml import Pipeline

# Definir las etapas del pipeline
pipeline = Pipeline(stages=[indexer, encoder, assembler, modelo_ml])

# Entrenar el modelo utilizando el pipeline
modelo_pipeline = pipeline.fit(train_data)

El uso de pipelines facilita la gestión de las etapas de transformación y modelado, y mejora la reproducibilidad del proceso.

Paso 9: Guardar y cargar el modelo

Persistir el modelo entrenado para su uso futuro:

# Guardar el modelo
modelo_pipeline.save('modelo_ml_pipeline')

# Cargar el modelo
from pyspark.ml.pipeline import PipelineModel
modelo_cargado = PipelineModel.load('modelo_ml_pipeline')

Esto es esencial para despliegues en producción y para evitar reentrenamientos innecesarios.

Paso 10: Optimización y tuning del modelo

Ajustar el modelo utilizando CrossValidator y una rejilla de hiperparámetros:

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator

# Definir la rejilla de hiperparámetros
paramGrid = ParamGridBuilder() \
    .addGrid(modelo_ml.regParam, [0.01, 0.1]) \
    .addGrid(modelo_ml.elasticNetParam, [0.0, 0.5, 1.0]) \
    .build()

# Configurar el CrossValidator
cv = CrossValidator(estimator=pipeline, estimatorParamMaps=paramGrid, evaluator=evaluador, numFolds=3)

# Entrenar el modelo con validación cruzada
modelo_cv = cv.fit(train_data)

# Evaluar el mejor modelo
mejores_predicciones = modelo_cv.bestModel.transform(test_data)
mejor_auc = evaluador.evaluate(mejores_predicciones)
print(f'Mejor AUC después del tuning: {mejor_auc}')

El tuning permite encontrar la mejor combinación de hiperparámetros, mejorando el rendimiento del modelo.

La migración del modelo a MLlib ofrece varios beneficios:

• Procesamiento en paralelo: Aprovecha la infraestructura distribuida de Spark.
• Manejo de Big Data: Capacidad para procesar datasets que no caben en memoria.
• Integración con Spark Ecosystem: Fácil integración con Spark SQL y Spark Streaming.
• Automatización: Uso de pipelines para flujos de trabajo más eficientes.