Puntos clave: El uso de Python con Spark ML facilita el desarrollo de modelos de clasificación, especialmente para predecir condiciones de salud como defectos cardíacos. El tutorial describe métodos para la preparación, transformación e implementación de regresión logística de datos.
- Aprovecha el DataFrame de Spark para gestionar de forma eficiente grandes conjuntos de datos con bibliotecas en Python.
- Recorre todo el flujo de trabajo, logrando una precisión del modelo del 93%.
Hemos estado escribiendo sobre cómo utilizar Spark ML con el lenguaje de programación Scala. Pero no muchos programadores conocen Scala. Python ha adelantado a Java en cuanto a número de usuarios, en gran parte gracias a la fuerza del aprendizaje automático. Así que, centremos nuestra atención en usar Spark ML con Python.
te podría decir que Spark están centrado en Scala. Scala tiene interfaces tanto de Python como de Scala e intérpretes de línea de comandos. Scala están el predeterminado. El de Python están llama pyspark. La mayoría de los ejemplos que da Spark son en Scala y, en algunos casos, ningún ejemplo son en Python.
Apache Atom
Python están el lenguaje preferido para utilizar en ciencia de datos debido a NumPy, Pandas y matplotlib, que son herramientas que Haz trabajar con arrays y dibujar gráficos más fácil y que pueden trabajar con grandes arrays de datos de forma eficiente. Pero Spark está diseñado para trabajar con una enorme cantidad de datos, distribuidos en un clúster. Es buena práctica utilizar ambas herramientas, cambiando de un lado a otro, quizás, según lo requiera la demanda.
Pero, como Ve, dado que el dataframe de Spark no están lo mismo que un dataframe de Pandas, no están compatibilidad del 100% entre todos estos objetos. te debe convertir dataframes de Spark en listas, arrays y otras estructuras para poder graficarlos con matplotlib. Como no te puede cortar arrays usando el familiar [:,4], se necesita más código para hacer la misma operación.
Pero el otro problema están el rendimiento. Apache Atom existe para convertir objetos en procesos Java de forma eficiente a procesos Python y viceversa. Spark está escrito en Java y Scala. Scala se apoya en Java. Python, por supuesto, se ejecuta en un proceso Python.
Arrow acelera las operaciones como la conversión de dataframes Spark a dataframes Pandas y con operaciones columna por columna como . withcolumn().
Spark comenta algunos de los problemas relacionados con esto y el cambio de configuración te necesita Haz en Spark para aprovechar este aumento de rendimiento en su documentación de Apache Arrow.
Datos de pacientes cardíacos
Descarga los datos del conjunto de datos de la Universidad de São Paulo, disponible aquí. Si te son curiosidad, Ve esta discusión.
Las columnas son:
- Edad
- Sexo
- Tipo de dolor torácico (4 valores)
- Presión arterial en reposo
- Colesterol sérico en mg/dl
- Glucosa en ayunas > 120 mg/dl
- Resultados electrocardiográficos en reposo (valores 0,1,2)
- Frecuencia cardíaca máxima alcanzada
- Angina inducida por el ejercicio
- Oldpeak = depresión ST inducida por el ejercicio en relación con el reposo
- Pendiente del segmento ST de ejercicio pico
- Número de vasos principales (0-3) coloreados por fluoroscopía
- Thal: 3 = normal; 6 = defecto fijo; 7 = defecto reversible
El código, explicado
El objetivo es construir un modelo predictivo de regresión logística binaria usando Spark ML y Python que prediga si alguien tiene un defecto cardíaco. El código que sigue está disponible en un cuaderno de Zeppelin aquí.
Primero, leemos los datos en y asignamos nombres de columnas. Como los datos están pequeños y porque Pandas están más fáciles, los leemos en un dataframe de Pandas. Luego los convertimos en un dataframe de Spark con spark.createDataFrame().
te puede Ve a lo que me refiero con que el dataframe de Spark carece de algunas de las características de Pandas. En particular, utiliza Pandas para poder utiliza . iloc() para tomar las primeras 13 columnas y eliminar la última, que parece ser ruido que no está destinado a los datos.
%spark.pyspark
import pandas as pd
from pyspark.sql.types import StructType, StructField, NumericType
cols = ('age',
'sex',
'chest pain',
'resting blood pressure',
'serum cholesterol',
'fasting blood sugar',
'resting electrocardiographic results',
'maximum heart rate achieved',
'exercise induced angina',
'ST depression induced by exercise relative to rest',
'the slope of the peak exercise ST segment',
'number of major vessels ',
'thal',
'last')
data = pd.read_csv('/home/ubuntu/Downloads/heart.csv', delimiter=' ', names=cols)
data = data.iloc[:,0:13]
data['isSick'] = data['thal'].apply(isSick)
df = spark.createDataFrame(data)
El tal de campo indica si el paciente tiene un problema cardíaco. Los números son los siguientes:
- Un valor de 3 significa que el paciente está sano (normal).
- Un valor de 6 significa que el problema de salud del paciente ha sido solucionado.
- Un valor de 7 significa que el problema de salud del paciente puede solucionarse.
Así que escribe esta función isSick() para marcar 0 como negativo y 1 como positivo, porque la regresión logística binaria requiere uno de dos resultados.
def isSick(x):
if x in (3,7):
return 0
else:
return 1
Con aprendizaje automático y problemas de clasificación o regresión tenemos:
- Una matriz de características, incluyendo la edad del paciente, el nivel de azúcar en sangre, etc.
- Un vector de etiquetas, que indica si el paciente tiene un problema cardíaco.
Como son usamos un cuaderno Zeppelin, y PySpark está el shell de comandos en Python para Spark, escribimos %spark.pyspark en la parte superior de cada celda de Zeppelin para indicar el lenguaje y el intérprete que queremos usar.
A continuación, indicamos qué columnas del dataframe df queremos utilizar como features. Luego utilizamos el VectorAasembler para poner los doce campos en una nueva columna llamada features que contiene todo esto como un array.
Ahora Crea el dataframe de Spark raw_data usando la operación transform() y seleccionando solo la columna de características.
%spark.pyspark
from pyspark.ml.feature import StandardScaler
from pyspark.ml.feature import VectorAssembler
features = ('age',
'sex',
'chest pain',
'resting blood pressure',
'serum cholestoral',
'fasting blood sugar',
'resting electrocardiographic results',
'maximum heart rate achieved',
'exercise induced angina',
'ST depression induced by exercise relative to rest',
'the slope of the peak exercise ST segment',
'number of major vessels ')
assembler = VectorAssembler(inputCols=features,outputCol="features")
raw_data=assembler.transform(df)
raw_data.select("features").show(truncate=False)
Utilizamos el Escalador Estándar para poner todos los números en la misma escala, lo cual es práctica estándar en aprendizaje automático. Esto toma la observación y resta la media, y luego la divide por la desviación estándar.
%spark.pyspark
from pyspark.ml.feature import StandardScaler
standardscaler=StandardScaler().setInputCol("features").setOutputCol("Scaled_features")
raw_data=standardscaler.fit(raw_data).transform(raw_data)
raw_data.select("features","Scaled_features").show(5)
Así están ahora los datos de características:
+--------------------------------------------------------+
|features |
+--------------------------------------------------------+
|[70.0,1.0,4.0,130.0,322.0,0.0,2.0,109.0,0.0,2.4,2.0,3.0]|
|[67.0,0.0,3.0,115.0,564.0,0.0,2.0,160.0,0.0,1.6,2.0,0.0]|
|[57.0,1.0,2.0,124.0,261.0,0.0,0.0,141.0,0.0,0.3,1.0,0.0]|
|[64.0,1.0,4.0,128.0,263.0,0.0,0.0,105.0,1.0,0.2,2.0,1.0]|
|[74.0,0.0,2.0,120.0,269.0,0.0,2.0,121.0,1.0,0.2,1.0,1.0]|
|[65.0,1.0,4.0,120.0,177.0,0.0,0.0,140.0,0.0,0.4,1.0,0.0]|
|[56.0,1.0,3.0,130.0,256.0,1.0,2.0,142.0,1.0,0.6,2.0,1.0]|
|[59.0,1.0,4.0,110.0,239.0,0.0,2.0,142.0,1.0,1.2,2.0,1.0]|
|[60.0,1.0,4.0,140.0,293.0,0.0,2.0,170.0,0.0,1.2,2.0,2.0]|
|[63.0,0.0,4.0,150.0,407.0,0.0,2.0,154.0,0.0,4.0,2.0,3.0]|
|[59.0,1.0,4.0,135.0,234.0,0.0,0.0,161.0,0.0,0.5,2.0,0.0]|
|[53.0,1.0,4.0,142.0,226.0,0.0,2.0,111.0,1.0,0.0,1.0,0.0]|
Como de costumbre, dividimos los datos en conjuntos de datos de entrenamiento y de prueba. No tenemos muchos datos, así que utilizamos una división 50/50.
%spark.pyspark
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit
training, test = raw_data.randomSplit([0.5, 0.5], seed=12345)
Ahora Crea el Modelo de Regresión Logística y lo entrenamos, es decir, que el modelo calcule los coeficientes y la interceptación que más se acerquen a los resultados que tenemos en la columna de etiquetas esSick
%spark.pyspark
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(labelCol="isSick", featuresCol="Scaled_features",maxIter=10)
model=lr.fit(training)
predict_train=model.transform(training)
predict_test=model.transform(test)
predict_test.select("isSick","prediction").show(10)
Aquí mostramos las primeras filas en comparación lado a lado. Estas son, en su mayoría, correctas.
+------+----------+
|isSick|prediction|
+------+----------+
| 0| 0.0|
| 1| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 1.0|
| 0| 1.0|
| 0| 0.0|
+------+----------+
Esto muestra los coeficientes y la intercepción.
Aquí son:
%spark.pyspark
print("Multinomial coefficients: " + str(model.coefficientMatrix))
print("Multinomial intercepts: " + str(model.interceptVector))
Aquí son:
Multinomial coefficients: DenseMatrix([[-0.41550466, 1.21573123, 0.16600826, 0.36478609, 0.33716549,
-0.020686 , -0.2092542 , -0.86514924, 0.1427418 , -0.3610241 ,
0.57324392, 0.42563706]])
Multinomial intercepts: [-0.2767309778166021]
Ahora utilizamos algunas funciones SQL de Spark F para Crea una nueva columna correcta cuando IsSick está igual a predicción, es decir, el resultado predicho es igual a los resultados reales.
%spark.pyspark
import pyspark.sql.functions as F
check = predict_test.withColumn('correct', F.when(F.col('isSick') == F.col('prediction'), 1).otherwise(0))
check.groupby("correct").count().show()
Aquí son los resultados:
+-------+-----+
|correct|count|
+-------+-----+
| 1| 137|
| 0| 10|
+-------+-----+
Así que, la precisión están 137 / 137 + 10 = 93%
Hay son otras formas de mostrar la precisión del modelo, como el área bajo la curva. Pero esta están la más sencilla de entender, a menos que te son un científico de datos y estadístico experimentado. Explicaremos formas más complejas de comprobar la precisión en futuros artículos.
Guía Apache Spark
- La Canalización de Aprendizaje Automático de Apache Spark: Una introducción
- Ejemplo de regresión lineal SGD con Apache Spark
- Clasificador de árbol de decisión de chispa
- Uso de regresión logística, scala y chispa
- Cómo utilizar cuadernos Jupyter con Apache Spark
- Uso de aprendizaje automático en Python y Spark para hacer clasificación
- Cómo escribir UDFs (funciones definidas por el usuario) de Spark en Python
- Ejemplo de Python Spark ML K-Means
- Cómo aplicar el aprendizaje automático a la ciberseguridad
- Apache Spark: Trabajando con Streams
- K-significa agrupación con Apache Spark
- Usando Chispa con Colmena
- Cómo utilizar Apache Spark para Haz predicciones para el mantenimiento preventivo
Estas publicaciones son mías y no representan necesariamente la posición, estrategias u opinión de BMC.