PySpark Tutorial para principiantes: aprende con EJEMPLOS
evelyn clarke
⚡ Resumen inteligente
PySpark son los Python API para Apache Spark, permitiendo a los analistas ejecutar procesamiento de datos distribuido, consultas SQL y aprendizaje automático en clústeres. Este tutorial lo instala y explica SparkContexto y SQLContext, luego construye un clasificador de regresión logística.
Antes de aprender PySpark, entendamos:
que es apache Spark?
Spark es una solución de big data que ha demostrado ser más fácil y rápida que Hadoop MapReduce. Spark es un software de código abierto desarrollado por el laboratorio RAD de UC Berkeley en 2009. Desde su lanzamiento al público en 2010, Spark ha ganado popularidad y se utiliza en la industria a una escala sin precedentes.
En la era del Big Data, los profesionales necesitan más que nunca herramientas rápidas y confiables para procesar flujos de datos. Herramientas anteriores como MapReduce eran las favoritas, pero eran lentas. Para superar este problema, Spark ofrece una solución rápida y de uso general. La principal diferencia entre Spark y MapReduce es que Spark Ejecuta cálculos en la memoria durante el transcurso del tiempo en el disco duro. Permite el acceso y procesamiento de datos a alta velocidad, reduciendo los tiempos de horas a minutos.
¿Qué es Py?Spark?
PySpark es una herramienta creada por Apache Spark Comunidad para usar Python con Spark. Permite trabajar con RDD (Resilient Distributed Dataset) en Python. También ofrece PySpark Shell para vincular Python API con Spark núcleo para iniciar Spark Contexto. Spark es el nombre del motor para realizar la computación en clúster, mientras que PySpark is Pythonbiblioteca para usar Spark.
La biblioteca se envía como la pyspark paquete en PyPI, por lo que un solo pip install pyspark incorpora el motor Scala/JVM más el puente Py4J que permite Python Los objetos lo utilizan como interfaz. Todo lo que sigue se basa en ese paquete.
Cómo Se Compara Spark funciona?
Spark Se basa en un motor computacional, lo que significa que se encarga de la programación, distribución y monitorización de la aplicación. Cada tarea se realiza en varias máquinas de trabajo llamadas clúster de computación. Un clúster de computación se refiere a la división de tareas. Una máquina realiza una tarea, mientras que las demás contribuyen al resultado final a través de una tarea diferente. Al final, todas las tareas se agregan para producir un resultado. Spark El administrador ofrece una descripción general de 360 grados de varios Spark Empleo.
El siguiente diagrama muestra cómo un programa controlador distribuye el trabajo a las máquinas trabajadoras que componen dicho clúster.
¿Cómo Spark Portafolio
Spark está diseñado para trabajar con
- Python
- Java
- Scala
- SQL
Una característica significativa de Spark es la gran cantidad de bibliotecas integradas, incluyendo MLlib para aprendizaje automático. Spark También está diseñado para funcionar con clústeres Hadoop y puede leer una amplia variedad de archivos, incluidos datos Hive, CSV, JSON, Cassandra datos entre otros.
¿Por qué utilizar Spark?
Como futuro profesional de datos, deberías estar familiarizado con PythonLas famosas bibliotecas de Pandas y scikit-aprenderEstas dos bibliotecas son fantásticas para explorar conjuntos de datos de tamaño medio. Los proyectos habituales de aprendizaje automático se basan en la siguiente metodología:
- Cargar los datos al disco.
- Importar los datos a la memoria de la máquina.
- Procesar/analizar los datos
- Construya el modelo de aprendizaje automático
- Almacenar la predicción en el disco
El problema surge cuando el científico de datos quiere procesar datos demasiado grandes para un solo ordenador. En los inicios de la ciencia de datos, los profesionales solían muestrear los datos, ya que no siempre era necesario entrenar con conjuntos de datos enormes. El científico de datos encontraba una buena muestra estadística, realizaba una comprobación de robustez adicional y obtenía un modelo excelente.
Sin embargo, existen algunos problemas con esto:
- ¿El conjunto de datos refleja el mundo real?
- ¿Los datos incluyen un ejemplo específico?
- ¿Es el modelo apto para el muestreo?
Tomemos como ejemplo las recomendaciones de los usuarios. Los recomendadores se basan en comparar usuarios con otros usuarios para evaluar sus preferencias. Si el profesional de datos toma solo un subconjunto de datos, no habrá una cohorte de usuarios que sean muy similares entre sí. Los recomendadores deben ejecutarse en todo el conjunto de datos o no ejecutarse en absoluto.
¿Cuál es la solución?
La solución ha sido evidente durante mucho tiempo: dividir el problema en varias computadoras. La computación paralela también conlleva múltiples problemas. Los desarrolladores a menudo tienen problemas para escribir código paralelo y terminan teniendo que resolver una serie de problemas complejos relacionados con el multiprocesamiento en sí.
PySpark Proporciona al científico de datos una API que puede utilizarse para resolver problemas de procesamiento de datos en paralelo. PySpark Gestiona las complejidades del procesamiento paralelo, como la distribución de datos, la distribución de código y la recopilación de la salida de los procesos en un clúster de máquinas.
Spark Puede ejecutarse de forma independiente, pero la mayoría de las veces se ejecuta sobre un marco de computación en clúster como Hadoop. Sin embargo, en pruebas y desarrollo, un científico de datos puede ejecutar de manera eficiente Spark en sus cajas de desarrollo o portátiles sin clúster
- Una de las principales ventajas de Spark Se trata de construir una arquitectura que abarque la gestión de la transmisión de datos, consultas de datos sin problemas, predicción de aprendizaje automático y acceso en tiempo real a diversos análisis.
- Spark trabaja en estrecha colaboración con SQL lenguaje, es decir, datos estructurados. Permite consultar los datos en tiempo real.
- La principal función de un científico de datos es analizar y crear modelos predictivos. En resumen, un científico de datos debe saber cómo consultar datos mediante SQL, generar informes estadísticos y utilizar el aprendizaje automático para realizar predicciones. Dedica una cantidad considerable de tiempo a limpiar, transformar y analizar los datos. Una vez que el conjunto de datos o el flujo de trabajo está listo, utiliza diversas técnicas para descubrir información valiosa y patrones ocultos. La manipulación de datos debe ser robusta y, a la vez, fácil de usar. Spark es la herramienta adecuada gracias a su velocidad y sus ricas API.
En este PySpark tutorial, aprenderá cómo construir un clasificador con PySpark ejemplos.
Cómo instalar PySpark con AWS
El Jupyter El equipo crea una imagen de Docker para ejecutar. Spark eficientemente. A continuación se detallan los pasos que puede seguir para instalar PySpark instancia en AWS.
Consulte nuestro tutorial sobre AWS y TensorFlow
Paso 1: crear una instancia
En primer lugar, necesitas crear una instancia. Vaya a su cuenta de AWS e inicie la instancia. Puede aumentar el almacenamiento hasta 15 gy utilizar el mismo grupo de seguridad que en el tutorial de TensorFlow.
Paso 2: abre la conexión
Abra la conexión e instale el contenedor Docker. Para obtener más detalles, consulte el tutorial con TensorFlow con Docker. Tenga en cuenta que debe estar en el directorio de trabajo correcto.
Simplemente ejecute estos códigos para instalar Docker:
sudo yum update -y sudo yum install -y docker sudo service docker start sudo user-mod -a -G docker ec2-user exit
Paso 3: Vuelva a abrir la conexión e instale Spark
Después de volver a abrir la conexión, puede instalar la imagen que contiene PySpark.
## Spark docker run -v ~/work:/home/jovyan/work -d -p 8888:8888 jupyter/pyspark-notebook ## Allow preserving Jupyter notebook sudo chown 1000 ~/work ## Install tree to see our working directory next sudo yum install -y tree
Paso 4: abierto Jupyter
Consulta el contenedor y su nombre.
docker ps
Inicie la ventana acoplable con los registros de la ventana acoplable seguidos del nombre de la ventana acoplable. Por ejemplo, Docker registra zealous_goldwasser
Vaya a su navegador y ejecute Jupyter. La dirección es http://localhost:8888/. Pega la contraseña proporcionada por el terminal.
Nota: Si desea cargar/descargar un archivo en su máquina AWS, puede utilizar el software. Cyberduck.
Cómo instalar PySpark on Windows/Mac con Conda
A continuación se muestra un proceso detallado sobre cómo instalar PySpark on Windows/Mac usando Anaconda:
Para instalar Spark en su máquina local, una práctica recomendada es crear un nuevo entorno conda. Este nuevo entorno se instalará Python 3.6, Spark y todas las dependencias.
Usuario de Mac
cd anaconda3 touch hello-spark.yml vi hello-spark.yml
Windows El sistema de reservas de escritorios, interactivo y fácil de usar, ayuda a gestores y empresas a adaptarse a la nueva rutina laboral. El sistema inteligente optimiza espacios y horarios según necesidades reales.
cd C:\Users\Admin\Anaconda3 echo.>hello-spark.yml notepad hello-spark.yml
Puede editar el archivo .yml. Tenga cuidado con la sangría. Se requieren dos espacios antes –
name: hello-spark
dependencies:
- python=3.6
- jupyter
- ipython
- numpy
- numpy-base
- pandas
- py4j
- pyspark
- pytz
Guárdalo y crea el entorno. lleva algo de tiempo
conda env create -f hello-spark.yml
Para obtener más detalles sobre la ubicación, consulte el tutorial Instalar TensorFlow
Puede comprobar todo el entorno instalado en su máquina.
conda env list
Activate hello-spark
Usuario de Mac
source activate hello-spark
Windows El sistema de reservas de escritorios, interactivo y fácil de usar, ayuda a gestores y empresas a adaptarse a la nueva rutina laboral. El sistema inteligente optimiza espacios y horarios según necesidades reales.
activate hello-spark
Nota: Ya ha creado un entorno de TensorFlow específico para ejecutar los tutoriales en TensorFlow. Es más conveniente crear un nuevo entorno diferente a hello-tf. No tiene sentido sobrecargar hello-tf con Spark o cualquier otra biblioteca de aprendizaje automático.
Imagine que la mayor parte de su proyecto involucra TensorFlow, pero necesita usar Spark para un proyecto en particular. Puede configurar un entorno TensorFlow para todo su proyecto y crear un entorno separado para Spark. Puede agregar tantas bibliotecas en Spark entorno como desee sin interferir con el entorno de TensorFlow. Una vez que haya terminado con el Sparkdel proyecto, puedes borrarlo sin afectar el entorno de TensorFlow.
Jupyter
Abra Jupyter Cuaderno y prueba si PySpark Funciona. En un nuevo cuaderno pegue el siguiente PySpark Código de muestra:
import pyspark from pyspark import SparkContext sc =SparkContext()
Una ejecución exitosa no imprime ningún resultado, como se muestra en la captura de pantalla del cuaderno a continuación.
Si se muestra un error, es probable que Java No está instalado en su máquina. En Mac, abra la terminal y escriba java -version. Si hay una versión de Java, asegúrese de que sea la 1.8. Windows, vaya a Aplicación y verifique si hay una Java carpeta. Si hay un Java carpeta, verifique que Java 1.8 está instalado. Al momento de escribir esto, PySpark no es compatible con Java 9 y superiores.
Nota de versión: que Java 8 restricciones se aplican a la Spark Línea 2.x utilizada a lo largo de este tutorial. Actual Spark Las versiones 3.x y 4.x se ejecutan en Java 8, 11 y 17, por lo que un JDK más reciente está bien si instalas un Py compatible.Spark versión.
Si necesita instalar Java, tu a pensar este enlace y descargar jdk-8u181-windows-x64.exe
Para usuarios de Mac, se recomienda utilizar "brew".
brew tap caskroom/versions brew cask install java8
Consulte este tutorial paso a paso sobre cómo instalar Java
Nota: Utilice remover para borrar un entorno por completo.
conda env remove -n hello-spark -y
Spark Contexto
SparkEl contexto es el motor interno que permite las conexiones con los clústeres. Si desea ejecutar una operación, necesita un SparkContexto.
Créar un SparkContexto
En primer lugar, es necesario iniciar un SparkContexto.
import pyspark from pyspark import SparkContext sc =SparkContext()
Ahora que la SparkEl contexto está listo, puedes crear una colección de datos llamada RDD, Resilient Distributed Dataset. El cálculo en un RDD se paraleliza automáticamente en todo el clúster.
nums= sc.parallelize([1,2,3,4])
Puedes acceder a la primera fila con tomar
nums.take(1)
[1]
Puede aplicar una transformación a los datos con una función lambda. en el pySpark En el siguiente ejemplo, se devuelve el cuadrado de números. Es una transformación del mapa.
squared = nums.map(lambda x: x*x).collect() for num in squared: print('%i ' % (num))
1 4 9 16
Contexto SQL
Una forma más conveniente es utilizar DataFrame. SparkEl contexto ya está configurado, puedes usarlo para crear el DataFrame. También necesitas declarar el SQLContext.
SQLContext permite conectar el motor con diferentes fuentes de datos. Se utiliza para iniciar las funcionalidades de Spark SQL
Nota de versión: SQLContext fue reemplazado por SparkSesión en Spark 2.0 y formalmente obsoleto en Spark 3.0. Los ejemplos a continuación mantienen SQLContext porque es en lo que se ejecuta el tutorial original; en un clúster moderno, el punto de entrada equivalente es SparkSession.builder.getOrCreate(), que expone los mismos métodos read y createDataFrame.
from pyspark.sql import Row from pyspark.sql import SQLContext sqlContext = SQLContext(sc)
Ahora en esto Spark tutoriales Python, creemos una lista de tuplas. Cada tupla contendrá el nombre de las personas y su edad. Se requieren cuatro pasos:
Paso 1) Crea la lista de tuplas con la información.
[('John',19),('Smith',29),('Adam',35),('Henry',50)]
Paso 2) Construir un RDD
rdd = sc.parallelize(list_p)
Paso 3) Convertir las tuplas
rdd.map(lambda x: Row(name=x[0], age=int(x[1])))
Paso 4) Crear un contexto de marco de datos
sqlContext.createDataFrame(ppl) list_p = [('John',19),('Smith',29),('Adam',35),('Henry',50)] rdd = sc.parallelize(list_p) ppl = rdd.map(lambda x: Row(name=x[0], age=int(x[1]))) DF_ppl = sqlContext.createDataFrame(ppl)
Si desea acceder al tipo de cada característica, puede usar printSchema()
DF_ppl.printSchema() root |-- age: long (nullable = true) |-- name: string (nullable = true)
Ejemplo de aprendizaje automático con PySpark
Ahora que tienes una breve idea de Spark y SQLContext, está listo para crear su primer programa de aprendizaje automático.
A continuación se muestran los pasos para crear un programa de aprendizaje automático con PySpark:
- Paso 1) Operación básica con PySpark
- Paso 2) Preprocesamiento de datos
- Paso 3) Construya un canal de procesamiento de datos
- Paso 4) Construir el clasificador: logístico
- Paso 5) Entrenar y evaluar el modelo.
- Paso 6) Ajusta el hiperparámetro
En este PySpark Tutorial de aprendizaje automático: utilizaremos el conjunto de datos de adultos. El objetivo de este tutorial es aprender a usar Python.SparkPara obtener más información sobre el conjunto de datos, consulte este tutorial.
Tenga en cuenta que el conjunto de datos no es significativo y puede pensar que el cálculo lleva mucho tiempo. Spark está diseñado para procesar una cantidad considerable de datos. SparkSu rendimiento aumenta en relación con otras bibliotecas de aprendizaje automático a medida que aumenta el tamaño del conjunto de datos procesado.
Paso 1) Operación básica con PySpark
En primer lugar, debe inicializar el SQLContext que aún no se ha iniciado.
#from pyspark.sql import SQLContext url = "https://raw.githubusercontent.com/guru99-edu/R-Programming/master/adult_data.csv" from pyspark import SparkFiles sc.addFile(url) sqlContext = SQLContext(sc)
luego, puede leer el archivo cvs con sqlContext.read.csv. Usas inferSchema configurado en True para saber Spark para adivinar automáticamente el tipo de datos. De forma predeterminada, cambia a Falso.
df = sqlContext.read.csv(SparkFiles.get("adult_data.csv"), header=True, inferSchema= True)
Echemos un vistazo al tipo de datos.
df.printSchema() root |-- age: integer (nullable = true) |-- workclass: string (nullable = true) |-- fnlwgt: integer (nullable = true) |-- education: string (nullable = true) |-- education_num: integer (nullable = true) |-- marital: string (nullable = true) |-- occupation: string (nullable = true) |-- relationship: string (nullable = true) |-- race: string (nullable = true) |-- sex: string (nullable = true) |-- capital_gain: integer (nullable = true) |-- capital_loss: integer (nullable = true) |-- hours_week: integer (nullable = true) |-- native_country: string (nullable = true) |-- label: string (nullable = true)
Puedes ver los datos con show.
df.show(5, truncate = False)
+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+----------+--------------+-----+ |age|workclass |fnlwgt|education|education_num|marital |occupation |relationship |race |sex |capital_gain|capital_loss|hours_week|native_country|label| +---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+----------+--------------+-----+ |39 |State-gov |77516 |Bachelors|13 |Never-married |Adm-clerical |Not-in-family|White|Male |2174 |0 |40 |United-States |<=50K| |50 |Self-emp-not-inc|83311 |Bachelors|13 |Married-civ-spouse|Exec-managerial |Husband |White|Male |0 |0 |13 |United-States |<=50K| |38 |Private |215646|HS-grad |9 |Divorced |Handlers-cleaners|Not-in-family|White|Male |0 |0 |40 |United-States |<=50K| |53 |Private |234721|11th |7 |Married-civ-spouse|Handlers-cleaners|Husband |Black|Male |0 |0 |40 |United-States |<=50K| |28 |Private |338409|Bachelors|13 |Married-civ-spouse|Prof-specialty |Wife |Black|Female|0 |0 |40 |Cuba |<=50K| +---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+----------+--------------+-----+ only showing top 5 rows
Si no configuraste inferSchema en True, esto es lo que está sucediendo con el tipo. Todos están en cadena.
df_string = sqlContext.read.csv(SparkFiles.get("adult.csv"), header=True, inferSchema= False) df_string.printSchema() root |-- age: string (nullable = true) |-- workclass: string (nullable = true) |-- fnlwgt: string (nullable = true) |-- education: string (nullable = true) |-- education_num: string (nullable = true) |-- marital: string (nullable = true) |-- occupation: string (nullable = true) |-- relationship: string (nullable = true) |-- race: string (nullable = true) |-- sex: string (nullable = true) |-- capital_gain: string (nullable = true) |-- capital_loss: string (nullable = true) |-- hours_week: string (nullable = true) |-- native_country: string (nullable = true) |-- label: string (nullable = true)
Para convertir la variable continua al formato correcto, puede utilizar la refundición de las columnas. Puedes usar withColumn para saber Spark En qué columna operar la transformación.
# Import all from `sql.types` from pyspark.sql.types import * # Write a custom function to convert the data type of DataFrame columns def convertColumn(df, names, newType): for name in names: df = df.withColumn(name, df[name].cast(newType)) return df # List of continuous features CONTI_FEATURES = ['age', 'fnlwgt','capital_gain', 'education_num', 'capital_loss', 'hours_week'] # Convert the type df_string = convertColumn(df_string, CONTI_FEATURES, FloatType()) # Check the dataset df_string.printSchema() root |-- age: float (nullable = true) |-- workclass: string (nullable = true) |-- fnlwgt: float (nullable = true) |-- education: string (nullable = true) |-- education_num: float (nullable = true) |-- marital: string (nullable = true) |-- occupation: string (nullable = true) |-- relationship: string (nullable = true) |-- race: string (nullable = true) |-- sex: string (nullable = true) |-- capital_gain: float (nullable = true) |-- capital_loss: float (nullable = true) |-- hours_week: float (nullable = true) |-- native_country: string (nullable = true) |-- label: string (nullable = true) from pyspark.ml.feature import StringIndexer #stringIndexer = StringIndexer(inputCol="label", outputCol="newlabel") #model = stringIndexer.fit(df) #df = model.transform(df) df.printSchema()
Seleccionar columnas
Puede seleccionar y mostrar las filas con select y los nombres de las funciones. A continuación, se seleccionan edad y fnlwgt.
df.select('age','fnlwgt').show(5)
+---+------+ |age|fnlwgt| +---+------+ | 39| 77516| | 50| 83311| | 38|215646| | 53|234721| | 28|338409| +---+------+ only showing top 5 rows
Contar por grupo
Si desea contar el número de ocurrencias por grupo, puede encadenar:
- agrupar por()
- contar()
juntos. en el pySpark En el siguiente ejemplo, se cuenta el número de filas según el nivel educativo.
df.groupBy("education").count().sort("count",ascending=True).show()
+------------+-----+ | education|count| +------------+-----+ | Preschool| 51| | 1st-4th| 168| | 5th-6th| 333| | Doctorate| 413| | 12th| 433| | 9th| 514| | Prof-school| 576| | 7th-8th| 646| | 10th| 933| | Assoc-acdm| 1067| | 11th| 1175| | Assoc-voc| 1382| | Masters| 1723| | Bachelors| 5355| |Some-college| 7291| | HS-grad|10501| +------------+-----+
Describe los datos
Para obtener un resumen estadístico de los datos, puede utilizar describe(). Calculará:
- contar
- mean
- desviación estándar
- min
- max
df.describe().show()
+-------+------------------+-----------+------------------+------------+-----------------+--------+----------------+------------+------------------+------+------------------+----------------+------------------+--------------+-----+ |summary| age| workclass| fnlwgt| education| education_num| marital| occupation|relationship| race| sex| capital_gain| capital_loss| hours_week|native_country|label| +-------+------------------+-----------+------------------+------------+-----------------+--------+----------------+------------+------------------+------+------------------+----------------+------------------+--------------+-----+ | count| 32561| 32561| 32561| 32561| 32561| 32561| 32561| 32561| 32561| 32561| 32561| 32561| 32561| 32561|32561| | mean| 38.58164675532078| null|189778.36651208502| null| 10.0806793403151| null| null| null| null| null|1077.6488437087312| 87.303829734959|40.437455852092995| null| null| | stddev|13.640432553581356| null|105549.97769702227| null|2.572720332067397| null| null| null| null| null| 7385.292084840354|402.960218649002|12.347428681731838| null| null| | min| 17| ?| 12285| 10th| 1|Divorced| ?| Husband|Amer-Indian-Eskimo|Female| 0| 0| 1| ?|<=50K| | max| 90|Without-pay| 1484705|Some-college| 16| Widowed|Transport-moving| Wife| White| Male| 99999| 4356| 99| Yugoslavia| >50K| +-------+------------------+-----------+------------------+------------+-----------------+--------+----------------+------------+------------------+------+------------------+----------------+------------------+--------------+-----+
Si desea el resumen estadístico de una sola columna, agregue el nombre de la columna dentro de describe()
df.describe('capital_gain').show()
+-------+------------------+ |summary| capital_gain| +-------+------------------+ | count| 32561| | mean|1077.6488437087312| | stddev| 7385.292084840354| | min| 0| | max| 99999| +-------+------------------+
Cálculo de tablas cruzadas
En algunas ocasiones, puede resultar interesante ver las estadísticas descriptivas entre dos columnas por pares. Por ejemplo, se puede contar la cantidad de personas con ingresos inferiores o superiores a 50 XNUMX dólares por nivel de educación. Esta operación se denomina tabla de referencias cruzadas.
df.crosstab('age', 'label').sort("age_label").show()
+---------+-----+----+ |age_label|<=50K|>50K| +---------+-----+----+ | 17| 395| 0| | 18| 550| 0| | 19| 710| 2| | 20| 753| 0| | 21| 717| 3| | 22| 752| 13| | 23| 865| 12| | 24| 767| 31| | 25| 788| 53| | 26| 722| 63| | 27| 754| 81| | 28| 748| 119| | 29| 679| 134| | 30| 690| 171| | 31| 705| 183| | 32| 639| 189| | 33| 684| 191| | 34| 643| 243| | 35| 659| 217| | 36| 635| 263| +---------+-----+----+ only showing top 20 rows
Se puede ver que ninguna persona tiene ingresos superiores a 50 cuando son jóvenes.
Columna de caída
Hay dos API intuitivas para eliminar columnas:
- soltar(): soltar una columna
- dropna(): Elimina NA
Debajo sueltas la columna num_educación
df.drop('education_num').columns ['age', 'workclass', 'fnlwgt', 'education', 'marital', 'occupation', 'relationship', 'race', 'sex', 'capital_gain', 'capital_loss', 'hours_week', 'native_country', 'label']
Filtrar datos
Puede utilizar filter() para aplicar estadísticas descriptivas en un subconjunto de datos. Por ejemplo, puedes contar el número de personas mayores de 40 años.
df.filter(df.age > 40).count()
13443
DescriptCinco estadísticas por grupo.
Finalmente, puedes agrupar datos por grupo y calcular operaciones estadísticas como la media.
df.groupby('marital').agg({'capital_gain': 'mean'}).show()
+--------------------+------------------+ | marital| avg(capital_gain)| +--------------------+------------------+ | Separated| 535.5687804878049| | Never-married|376.58831788823363| |Married-spouse-ab...| 653.9832535885167| | Divorced| 728.4148098131893| | Widowed| 571.0715005035247| | Married-AF-spouse| 432.6521739130435| | Married-civ-spouse|1764.8595085470085| +--------------------+------------------+
Paso 2) Preprocesamiento de datos
El procesamiento de datos es un paso crítico en el aprendizaje automático. Después de eliminar los datos basura, obtendrá información importante.
Por ejemplo, sabes que la edad no es una función lineal con los ingresos. Cuando las personas son jóvenes, sus ingresos suelen ser inferiores a los de la mediana edad. Después de la jubilación, un hogar utiliza sus ahorros, lo que significa una disminución de sus ingresos. Para capturar este patrón, puede agregar un cuadrado a la función de edad.
Añadir cuadro de edad
Para agregar una nueva función, debe:
- Seleccione la columna
- Aplicar la transformación y agregarla al DataFrame.
from pyspark.sql.functions import * # 1 Select the column age_square = df.select(col("age")**2) # 2 Apply the transformation and add it to the DataFrame df = df.withColumn("age_square", col("age")**2) df.printSchema() root |-- age: integer (nullable = true) |-- workclass: string (nullable = true) |-- fnlwgt: integer (nullable = true) |-- education: string (nullable = true) |-- education_num: integer (nullable = true) |-- marital: string (nullable = true) |-- occupation: string (nullable = true) |-- relationship: string (nullable = true) |-- race: string (nullable = true) |-- sex: string (nullable = true) |-- capital_gain: integer (nullable = true) |-- capital_loss: integer (nullable = true) |-- hours_week: integer (nullable = true) |-- native_country: string (nullable = true) |-- label: string (nullable = true) |-- age_square: double (nullable = true)
Puede ver que age_square se agregó correctamente al marco de datos. Puede cambiar el orden de las variables con select. A continuación, traes age_square justo después de edad.
COLUMNS = ['age', 'age_square', 'workclass', 'fnlwgt', 'education', 'education_num', 'marital', 'occupation', 'relationship', 'race', 'sex', 'capital_gain', 'capital_loss', 'hours_week', 'native_country', 'label'] df = df.select(COLUMNS) df.first()
Row(age=39, age_square=1521.0, workclass='State-gov', fnlwgt=77516, education='Bachelors', education_num=13, marital='Never-married', occupation='Adm-clerical', relationship='Not-in-family', race='White', sex='Male', capital_gain=2174, capital_loss=0, hours_week=40, native_country='United-States', label='<=50K')
Excluir Holanda-Países Bajos
Cuando un grupo dentro de una característica tiene solo una observación, no aporta información al modelo. Por el contrario, puede dar lugar a un error durante la validación cruzada.
Comprobemos el origen del hogar.
df.filter(df.native_country == 'Holand-Netherlands').count() df.groupby('native_country').agg({'native_country': 'count'}).sort(asc("count(native_country)")).show()
+--------------------+---------------------+ | native_country|count(native_country)| +--------------------+---------------------+ | Holand-Netherlands| 1| | Scotland| 12| | Hungary| 13| | Honduras| 13| |Outlying-US(Guam-...| 14| | Yugoslavia| 16| | Thailand| 18| | Laos| 18| | Cambodia| 19| | Trinadad&Tobago| 19| | Hong| 20| | Ireland| 24| | Ecuador| 28| | Greece| 29| | France| 29| | Peru| 31| | Nicaragua| 34| | Portugal| 37| | Iran| 43| | Haiti| 44| +--------------------+---------------------+ only showing top 20 rows
La característica país_nativo tiene solo un hogar procedente de Holanda. Lo excluyes.
df_remove = df.filter(df.native_country != 'Holand-Netherlands')
Paso 3) Construya un canal de procesamiento de datos
Similar a scikit-learn, PySpark tiene una API de canalización.
Una canalización es muy conveniente para mantener la estructura de los datos. Insertas los datos en la tubería. Dentro de la canalización, se realizan varias operaciones y la salida se utiliza para alimentar el algoritmo.
Por ejemplo, una transformación universal en el aprendizaje automático consiste en convertir una cadena en un codificador activo, es decir, una columna por grupo. Un codificador activo suele ser una matriz llena de ceros.
Los pasos para transformar los datos son muy similares a los de scikit-learn. Necesitas:
- Indexar la cadena a numérico
- Crea el codificador activo
- Transforma los datos
Dos API hacen el trabajo: StringIndexer, OneHotEncoder
- En primer lugar, selecciona la columna de cadena a indexar. inputCol es el nombre de la columna en el conjunto de datos. OutputCol es el nuevo nombre dado a la columna transformada.
StringIndexer(inputCol="workclass", outputCol="workclass_encoded")
- Ajustar los datos y transformarlos
model = stringIndexer.fit(df)
`indexed = model.transform(df)``
- Crea las columnas de noticias según el grupo. Por ejemplo, si hay 10 grupos en la función, la nueva matriz tendrá 10 columnas, una para cada grupo.
OneHotEncoder(dropLast=False, inputCol="workclassencoded", outputCol="workclassvec")
### Example encoder from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler stringIndexer = StringIndexer(inputCol="workclass", outputCol="workclass_encoded") model = stringIndexer.fit(df) indexed = model.transform(df) encoder = OneHotEncoder(dropLast=False, inputCol="workclass_encoded", outputCol="workclass_vec") encoded = encoder.transform(indexed) encoded.show(2)
+---+----------+----------------+------+---------+-------------+------------------+---------------+-------------+-----+----+------------+------------+----------+--------------+-----+-----------------+-------------+ |age|age_square| workclass|fnlwgt|education|education_num| marital| occupation| relationship| race| sex|capital_gain|capital_loss|hours_week|native_country|label|workclass_encoded|workclass_vec| +---+----------+----------------+------+---------+-------------+------------------+---------------+-------------+-----+----+------------+------------+----------+--------------+-----+-----------------+-------------+ | 39| 1521.0| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White|Male| 2174| 0| 40| United-States|<=50K| 4.0|(9,[4],[1.0])| | 50| 2500.0|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse|Exec-managerial| Husband|White|Male| 0| 0| 13| United-States|<=50K| 1.0|(9,[1],[1.0])| +---+----------+----------------+------+---------+-------------+------------------+---------------+-------------+-----+----+------------+------------+----------+--------------+-----+-----------------+-------------+ only showing top 2 rows
construir el oleoducto
Construirá una canalización para convertir todas las características precisas y agregarlas al conjunto de datos final. La canalización tendrá cuatro operaciones, pero siéntase libre de agregar tantas operaciones como desee.
- Codificar los datos categóricos.
- Indexar la característica de la etiqueta
- Agregar variable continua
- Montar los escalones.
Cada paso se almacena en una lista denominada etapas. Esta lista le indicará a VectorAssembler qué operación realizar dentro de la tubería.
1. Codifique los datos categóricos.
Este paso es exactamente igual que el ejemplo anterior, excepto que se itera sobre todas las características categóricas.
from pyspark.ml import Pipeline from pyspark.ml.feature import OneHotEncoderEstimator CATE_FEATURES = ['workclass', 'education', 'marital', 'occupation', 'relationship', 'race', 'sex', 'native_country'] stages = [] # stages in our Pipeline for categoricalCol in CATE_FEATURES: stringIndexer = StringIndexer(inputCol=categoricalCol, outputCol=categoricalCol + "Index") encoder = OneHotEncoderEstimator(inputCols=[stringIndexer.getOutputCol()], outputCols=[categoricalCol + "classVec"]) stages += [stringIndexer, encoder]
Nota sobre la API: OneHotEncoderEstimator se fusionó de nuevo con OneHotEncoder en Spark 3.0. En Spark En la versión 3.x o posterior, importe OneHotEncoder y pase los mismos argumentos inputCols y outputCols; el resto del bucle permanece sin cambios.
2. Indexar la característica de la etiqueta
Spark, como muchas otras bibliotecas, no acepta valores de cadena para la etiqueta. Convierte la función de etiqueta con StringIndexer y la agrega a las etapas de la lista
# Convert label into label indices using the StringIndexer label_stringIdx = StringIndexer(inputCol="label", outputCol="newlabel") stages += [label_stringIdx]
3. Agregar variable continua
La columna `inputCols` del `VectorAssembler` es una lista de columnas. Puedes crear una nueva lista que contenga todas las columnas nuevas. El código que aparece a continuación rellena la lista con características categóricas codificadas y características continuas.
assemblerInputs = [c + "classVec" for c in CATE_FEATURES] + CONTI_FEATURES
4. Ensamble los escalones.
Finalmente, pasas todos los pasos en el VectorAssembler.
assembler = VectorAssembler(inputCols=assemblerInputs, outputCol="features")stages += [assembler]
Ahora que todos los pasos están listos, envía los datos a la canalización.
# Create a Pipeline.
pipeline = Pipeline(stages=stages)
pipelineModel = pipeline.fit(df_remove)
model = pipelineModel.transform(df_remove)
Si revisa el nuevo conjunto de datos, puede ver que contiene todas las características, transformadas y no transformadas. Solo le interesa la nueva etiqueta y las características. Las características incluyen todas las características transformadas y las variables continuas.
model.take(1) [Row(age=39, age_square=1521.0, workclass='State-gov', fnlwgt=77516, education='Bachelors', education_num=13, marital='Never-married', occupation='Adm-clerical', relationship='Not-in-family', race='White', sex='Male', capital_gain=2174, capital_loss=0, hours_week=40, native_country='United-States', label='<=50K', workclassIndex=4.0, workclassclassVec=SparseVector(8, {4: 1.0}), educationIndex=2.0, educationclassVec=SparseVector(15, {2: 1.0}), maritalIndex=1.0, maritalclassVec=SparseVector(6, {1: 1.0}), occupationIndex=3.0, occupationclassVec=SparseVector(14, {3: 1.0}), relationshipIndex=1.0, relationshipclassVec=SparseVector(5, {1: 1.0}), raceIndex=0.0, raceclassVec=SparseVector(4, {0: 1.0}), sexIndex=0.0, sexclassVec=SparseVector(1, {0: 1.0}), native_countryIndex=0.0, native_countryclassVec=SparseVector(40, {0: 1.0}), newlabel=0.0, features=SparseVector(99, {4: 1.0, 10: 1.0, 24: 1.0, 32: 1.0, 44: 1.0, 48: 1.0, 52: 1.0, 53: 1.0, 93: 39.0, 94: 77516.0, 95: 2174.0, 96: 13.0, 98: 40.0}))]
Paso 4) Construir el clasificador: logístico
Para agilizar el cálculo, convierta el modelo en un DataFrame.
Debe seleccionar una nueva etiqueta y características del modelo usando el mapa.
from pyspark.ml.linalg import DenseVector input_data = model.rdd.map(lambda x: (x["newlabel"], DenseVector(x["features"])))
Está listo para crear los datos del tren como un DataFrame. Usas el sqlContext
df_train = sqlContext.createDataFrame(input_data, ["label", "features"])
Revisa la segunda fila
df_train.show(2)
+-----+--------------------+ |label| features| +-----+--------------------+ | 0.0|[0.0,0.0,0.0,0.0,...| | 0.0|[0.0,1.0,0.0,0.0,...| +-----+--------------------+ only showing top 2 rows
Crear un conjunto de tren/prueba
Divide el conjunto de datos 80/20 con randomSplit.
# Split the data into train and test sets
train_data, test_data = df_train.randomSplit([.8,.2],seed=1234)
Contemos cuántas personas con ingresos inferiores o superiores a 50 XNUMX tanto en el conjunto de entrenamiento como en el de prueba
train_data.groupby('label').agg({'label': 'count'}).show()
+-----+------------+ |label|count(label)| +-----+------------+ | 0.0| 19698| | 1.0| 6263| +-----+------------+
test_data.groupby('label').agg({'label': 'count'}).show()
+-----+------------+ |label|count(label)| +-----+------------+ | 0.0| 5021| | 1.0| 1578| +-----+------------+
Construir el regresor logístico
Por último, pero no menos importante, puedes construir el clasificador. PySpark Dispone de una API llamada LogisticRegression para realizar regresiones logísticas.
Inicializa lr indicando la columna de etiqueta y las columnas de características. Establece un máximo de 10 iteraciones y agrega un parámetro de regularización con un valor de 0.3. Tenga en cuenta que en la siguiente sección utilizará la validación cruzada con una cuadrícula de parámetros para ajustar el modelo.
# Import `LinearRegression` from pyspark.ml.classification import LogisticRegression # Initialize `lr` lr = LogisticRegression(labelCol="label", featuresCol="features", maxIter=10, regParam=0.3) # Fit the data to the model linearModel = lr.fit(train_data)
#Puedes ver los coeficientes de la regresión.
# Print the coefficients and intercept for logistic regression print("Coefficients: " + str(linearModel.coefficients)) print("Intercept: " + str(linearModel.intercept))
Coefficients: [-0.0678914665262,-0.153425526813,-0.0706009536407,-0.164057586562,-0.120655298528,0.162922330862,0.149176870438,-0.626836362611,-0.193483661541,-0.0782269980838,0.222667203836,0.399571096381,-0.0222024341804,-0.311925857859,-0.0434497788688,-0.306007744328,-0.41318209688,0.547937504247,-0.395837350854,-0.23166535958,0.618743906733,-0.344088614546,-0.385266881369,0.317324463006,-0.350518889186,-0.201335923138,-0.232878560088,-0.13349278865,-0.119760542498,0.17500602491,-0.0480968101118,0.288484253943,-0.116314616745,0.0524163478063,-0.300952624551,-0.22046421474,-0.16557996579,-0.114676231939,-0.311966431453,-0.344226119233,0.105530129507,0.152243047814,-0.292774545497,0.263628334433,-0.199951374076,-0.30329422583,-0.231087515178,0.418918551,-0.0565930184279,-0.177818073048,-0.0733236680663,-0.267972912252,0.168491215697,-0.12181255723,-0.385648075442,-0.202101794517,0.0469791640782,-0.00842850210625,-0.00373211448629,-0.259296141281,-0.309896554133,-0.168434409756,-0.11048086026,0.0280647963877,-0.204187030092,-0.414392623536,-0.252806580669,0.143366465705,-0.516359222663,-0.435627370849,-0.301949286524,0.0878249035894,-0.210951740965,-0.621417928742,-0.099445190784,-0.232671473401,-0.1077745606,-0.360429419703,-0.420362959052,-0.379729467809,-0.395186242741,0.0826401853838,-0.280251589972,0.187313505214,-0.20295228799,-0.431177064626,0.149759018379,-0.107114299614,-0.319314858424,0.0028450133235,-0.651220387649,-0.327918792207,-0.143659581445,0.00691075160413,8.38517628783e-08,2.18856717378e-05,0.0266701216268,0.000231075966823,0.00893832698698] Intercept: -1.9884177974805692
Paso 5) Entrenar y evaluar el modelo.
Para generar predicción para su conjunto de prueba,
Puedes usar linearModel con transform() en test_data
# Make predictions on test data using the transform() method.
predictions = linearModel.transform(test_data)
Puedes imprimir los elementos en las predicciones.
predictions.printSchema() root |-- label: double (nullable = true) |-- features: vector (nullable = true) |-- rawPrediction: vector (nullable = true) |-- probability: vector (nullable = true) |-- prediction: double (nullable = false)
Te interesa la etiqueta, la predicción y la probabilidad.
selected = predictions.select("label", "prediction", "probability") selected.show(20)
+-----+----------+--------------------+ |label|prediction| probability| +-----+----------+--------------------+ | 0.0| 0.0|[0.91560704124179...| | 0.0| 0.0|[0.92812140213994...| | 0.0| 0.0|[0.92161406774159...| | 0.0| 0.0|[0.96222760777142...| | 0.0| 0.0|[0.66363283056957...| | 0.0| 0.0|[0.65571324475477...| | 0.0| 0.0|[0.73053376932829...| | 0.0| 1.0|[0.31265053873570...| | 0.0| 0.0|[0.80005907577390...| | 0.0| 0.0|[0.76482251301640...| | 0.0| 0.0|[0.84447301189069...| | 0.0| 0.0|[0.75691912026619...| | 0.0| 0.0|[0.60902504096722...| | 0.0| 0.0|[0.80799228385509...| | 0.0| 0.0|[0.87704364852567...| | 0.0| 0.0|[0.83817652582377...| | 0.0| 0.0|[0.79655423248500...| | 0.0| 0.0|[0.82712311232246...| | 0.0| 0.0|[0.81372823882016...| | 0.0| 0.0|[0.59687710752201...| +-----+----------+--------------------+ only showing top 20 rows
Evaluar el modelo
Es necesario observar la métrica de precisión para ver qué tan bien (o mal) se desempeña el modelo. Actualmente, no existe una API para calcular la medida de precisión en Spark. El valor predeterminado es la curva característica operativa del receptor, ROC. Es una métrica diferente que tiene en cuenta la tasa de falsos positivos.
Antes de mirar la República de China, construyamos la medida de precisión. Estás más familiarizado con esta métrica. La medida de precisión es la suma de la predicción correcta sobre el número total de observaciones.
Creas un DataFrame con la etiqueta y la `predicción.
cm = predictions.select("label", "prediction")
Puedes consultar el número de clase en la etiqueta y la predicción.
cm.groupby('label').agg({'label': 'count'}).show()
+-----+------------+ |label|count(label)| +-----+------------+ | 0.0| 5021| | 1.0| 1578| +-----+------------+
cm.groupby('prediction').agg({'prediction': 'count'}).show()
+----------+-----------------+ |prediction|count(prediction)| +----------+-----------------+ | 0.0| 5982| | 1.0| 617| +----------+-----------------+
Por ejemplo, en el conjunto de prueba, hay 1578 hogares con ingresos superiores a 50 y 5021 por debajo. El clasificador, sin embargo, predijo 617 hogares con ingresos superiores a 50.
Puede calcular la precisión calculando el recuento cuando las etiquetas están clasificadas correctamente sobre el número total de filas.
cm.filter(cm.label == cm.prediction).count() / cm.count()
0.8237611759357478
Puede unir todo y escribir una función para calcular la precisión.
def accuracy_m(model): predictions = model.transform(test_data) cm = predictions.select("label", "prediction") acc = cm.filter(cm.label == cm.prediction).count() / cm.count() print("Model accuracy: %.3f%%" % (acc * 100)) accuracy_m(model = linearModel) Model accuracy: 82.376%
Métricas de la República de China
El módulo BinaryClassificationEvaluator incluye las medidas ROC. El receptor OperaLa curva característica es otra herramienta común utilizada con la clasificación binaria. Es muy similar a la curva de precisión/recuperación, pero en lugar de representar gráficamente la precisión versus la recuperación, la curva ROC muestra la tasa de verdaderos positivos (es decir, recuperación) frente a la tasa de falsos positivos. La tasa de falsos positivos es la proporción de casos negativos que se clasifican incorrectamente como positivos. Es igual a uno menos la tasa negativa verdadera. La tasa de verdaderos negativos también se llama especificidad. Por lo tanto, la curva ROC representa la sensibilidad (recuerdo) versus 1 – especificidad.
### Use ROC from pyspark.ml.evaluation import BinaryClassificationEvaluator # Evaluate model evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction") print(evaluator.evaluate(predictions)) print(evaluator.getMetricName())
0.8940481662695192áreaBajoROC
print(evaluator.evaluate(predictions))
0.8940481662695192
Paso 6) Ajusta el hiperparámetro
Por último, pero no menos importante, puedes ajustar los hiperparámetros. De forma similar a scikit-learn, creas una cuadrícula de parámetros y añades los que deseas ajustar.
Para reducir el tiempo de cálculo, solo ajusta el parámetro de regularización con solo dos valores.
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator # Create ParamGrid for Cross Validation paramGrid = (ParamGridBuilder() .addGrid(lr.regParam, [0.01, 0.5]) .build())
Finalmente, se evalúa el modelo mediante el método de validación cruzada con 5 pliegues. El entrenamiento tarda aproximadamente 16 minutos.
from time import * start_time = time() # Create 5-fold CrossValidator cv = CrossValidator(estimator=lr, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=5) # Run cross validations cvModel = cv.fit(train_data) # likely take a fair amount of time end_time = time() elapsed_time = end_time - start_time print("Time to train model: %.3f seconds" % elapsed_time)
Tiempo para entrenar el modelo: 978.807 segundos.
El mejor hiperparámetro de regularización es 0.01, con una precisión del 85.316 por ciento.
accuracy_m(model = cvModel) Model accuracy: 85.316%
Puedes extract el parámetro recomendado encadenando cvModel.bestModel con extractParamMap()
bestModel = cvModel.bestModel bestModel.extractParamMap()
{Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='aggregationDepth', doc='suggested depth for treeAggregate (>= 2)'): 2,
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty'): 0.0,
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='family', doc='The name of family which is a description of the label distribution to be used in the model. Supported options: auto, binomial, multinomial.'): 'auto',
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='featuresCol', doc='features column name'): 'features',
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='fitIntercept', doc='whether to fit an intercept term'): True,
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='labelCol', doc='label column name'): 'label',
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='maxIter', doc='maximum number of iterations (>= 0)'): 10,
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='predictionCol', doc='prediction column name'): 'prediction',
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='probabilityCol', doc='Column name for predicted class conditional probabilities. Note: Not all models output well-calibrated probability estimates! These probabilities should be treated as confidences, not precise probabilities'): 'probability',
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='rawPredictionCol', doc='raw prediction (a.k.a. confidence) column name'): 'rawPrediction',
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='regParam', doc='regularization parameter (>= 0)'): 0.01,
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='standardization', doc='whether to standardize the training features before fitting the model'): True,
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='threshold', doc='threshold in binary classification prediction, in range [0, 1]'): 0.5,
Param(parent='LogisticRegression_4d8f8ce4d6a02d8c29a0', name='tol', doc='the convergence tolerance for iterative algorithms (>= 0)'): 1e-06}
