<jupyter_start><jupyter_text>![logo](assets/logo.png "Logo") Módulo 4 - Big DataAutor: <NAME> - Prueba Steven  Entrenamiento de modelosDedicaremos esta lección para el desarrollo de un ejemplo de entrenamiento de un modelo básico, de principio a fin, en Apache Spark. Iniciaremos desde la lectura de los datos, limpieza de los mismos, análisis descriptivo, almacenamiento intermedio y entrenamiento de un modelo de regresión.Para esta actividad utilizaremos un conjunto de datos relacionado con la predicción del precio de viviendas, que se puede descargar desde https://www.kaggle.com/harlfoxem/housesalesprediction  Lectura y limpiezaComo hemos visto en módulos anteriores, una de las tareas más comunes al procesar datos es el ajuste y limpieza de los datos. Este ejemplo refleja esta situación. En las celdas siguientes leeremos los datos de un archivo CSV y realizaremos diferentes pasos para poder cargar los datos a una tabla en PostgreSQL que almacenará el conjunto de datos en su forma deseada para abordar el análisis.<jupyter_code># Cargar el conjunto de datos completo. Este paso no realiza ningún ajuste; simplemente lectura
import findspark
findspark.init('/usr/lib/python3.7/site-packages/pyspark')

from pyspark.sql.types import (StringType, IntegerType, FloatType, 
                               DecimalType, StructField, StructType)

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Basic JDBC pipeline") \
    .config("spark.driver.extraClassPath", "postgresql-42.2.14.jar") \
    .config("spark.executor.extraClassPath", "postgresql-42.2.14.jar") \
    .getOrCreate()

houses_df = spark \
    .read \
    .format("csv") \
    .option("path", "kc_house_data.csv") \
    .option("header", True) \
    .schema(StructType([
                StructField("id", StringType()),
                StructField("date", StringType()),
                StructField("price", StringType()),
                StructField("bedrooms", IntegerType()),
                StructField("bathrooms", FloatType()),
                StructField("sqft_living", IntegerType()),
                StructField("sqft_lot", IntegerType()),
                StructField("floors", StringType()),
                StructField("waterfront", IntegerType()),
                StructField("view", IntegerType()),
                StructField("condition", IntegerType()),
                StructField("grade", IntegerType()),
                StructField("sqft_above", IntegerType()),
                StructField("sqft_basement", IntegerType()),
                StructField("yr_built", IntegerType()),
                StructField("yr_renovated", IntegerType()),
                StructField("zipcode", StringType()),
                StructField("lat", FloatType()),
                StructField("long", FloatType()),
                StructField("sqft_living15", IntegerType()),
                StructField("sqft_lot15", IntegerType())])) \
    .load()

houses_df.printSchema()
houses_df.show()
# Seleccionar un subconjunto de las columnas de interés y el objectivo que es el precio
columns_kept = ['bedrooms', 'bathrooms', 'sqft_living', 'sqft_lot',
                'floors', 'condition', 'yr_built', 'price']
selected_columns_df = houses_df.select(columns_kept)

selected_columns_df.show()
# La cantidad de pisos y el precio fueron cargados como string. Deben ajustarse.
correct_types_df = selected_columns_df \
    .withColumn('floors', selected_columns_df['floors'].cast(FloatType())) \
    .withColumn('price', selected_columns_df['price'].cast(DecimalType()))


correct_types_df.show()
# Almacenar el conjunto de datos limpio en la base de datos
correct_types_df \
    .write \
    .format("jdbc") \
    .mode('overwrite') \
    .option("url", "jdbc:postgresql://host.docker.internal:5433/postgres") \
    .option("user", "postgres") \
    .option("password", "<PASSWORD>") \
    .option("dbtable", "houses") \
    .save()<jupyter_output><empty_output><jupyter_text>Inspección de datosPrevio a entrenar el modelo es común que se realice algún tipo de descripción de los datos, para tener una idea del tipo de problema con el que nos enfrentamos. A continuación, algunas operacions útiles que podemos usar.<jupyter_code># Cargar el conjunto de datos. Esta vez desde la base de datos

# Reading single DataFrame in Spark by retrieving all rows from a DB table.
df = spark \
    .read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host.docker.internal:5433/postgres") \
    .option("user", "postgres") \
    .option("password", "<PASSWORD>") \
    .option("dbtable", "houses") \
    .load()

df.show()
# Información descriptiva del dataframe.
df.describe(['bedrooms', 'floors', 'yr_built']).show()
# Para realizar operaciones más detalladas es necesario expresar las filas originales en vectores
from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(
    inputCols=['bedrooms', 'bathrooms', 'sqft_living', 'sqft_lot',
               'floors', 'condition', 'yr_built'],
    outputCol='features')

vector_df = assembler.transform(df)
vector_df = vector_df.select(['features', 'price'])
vector_df.show()

# Con la representación de vectores podemos calcular correlaciones
from pyspark.ml.stat import Correlation
import seaborn as sns
import matplotlib.pyplot as plt

pearson_matrix = Correlation.corr(vector_df, 'features').collect()[0][0]

sns.heatmap(pearson_matrix.toArray(), annot=True, fmt=".2f", cmap='viridis')<jupyter_output><empty_output><jupyter_text>EstandarizaciónComo recordamos de los módulos anteriores es deseable que los datos se encuentren estandarizados o normalizados, para evitar que la magnitud de ciertos atributos dominen el proceso de entrenamiento. El módulo `pyspark.ml.feature` tiene varias alternativas. A continuación se presenta una.<jupyter_code>from pyspark.ml.feature import StandardScaler

standard_scaler = StandardScaler(inputCol='features', outputCol='scaled')
scale_model = standard_scaler.fit(vector_df)

scaled_df = scale_model.transform(vector_df)
scaled_df.show()<jupyter_output>+--------------------+-------+--------------------+
|            features|  price|              scaled|
+--------------------+-------+--------------------+
|[3.0,1.0,1180.0,5...| 221900|[3.22559199779093...|
|[3.0,2.25,2570.0,...| 538000|[3.22559199779093...|
|[2.0,1.0,770.0,10...| 180000|[2.15039466519395...|
|[4.0,3.0,1960.0,5...| 604000|[4.30078933038790...|
|[3.0,2.0,1680.0,8...| 510000|[3.22559199779093...|
|[4.0,4.5,5420.0,1...|1225000|[4.30078933038790...|
|[3.0,2.25,1715.0,...| 257500|[3.22559199779093...|
|[3.0,1.5,1060.0,9...| 291850|[3.22559199779093...|
|[3.0,1.0,1780.0,7...| 229500|[3.22559199779093...|
|[3.0,2.5,1890.0,6...| 323000|[3.22559199779093...|
|[3.0,2.5,3560.0,9...| 662500|[3.22559199779093...|
|[2.0,1.0,1160.0,6...| 468000|[2.15039466519395...|
|[3.0,1.0,1430.0,1...| 310000|[3.22559199779093...|
|[3.0,1.75,1370.0,...| 400000|[3.22559199779093...|
|[5.0,2.0,1810.0,4...| 530000|[5.37598666298488...|
|[4.0,3.0,2950.0,5...| 650000|[4.30078933038790...|
|[3.0,2.0,18[...]<jupyter_text>Regresión linealComo ejemplo de un tipo de modelo a entrenar usaremos una regresión lineal básica para tratar de predecir el precio. Para ellos utilizaremos los datos estandarizados.<jupyter_code># Nótese que no se hace partición de datos de entrenamiento (ejercicio posterior).
from pyspark.ml.regression import LinearRegression

regression = LinearRegression(featuresCol='scaled', labelCol='price')
regression_model = regression.fit(scaled_df)

print('Pesos: {}\n b: {}'.format(regression_model.coefficients, regression_model.intercept))

print('RMSE: {} r2: {}'.format(
    regression_model.summary.rootMeanSquaredError,
    regression_model.summary.r2))

scaled_df.describe().show()

# Si se quiere hacer predicciones nada más se requiere llamar a regression_model.transform(df)
# Esto regresa el dataset con una columna nueva<jupyter_output>Pesos: [-64395.47318687582,50862.92091861797,278511.11197487527,-12339.081074522004,30946.977799688528,11683.836709724465,-94825.53343750679]
 b: 6223698.139208419
RMSE: 244323.53190293963 r2: 0.5570878698871067
+-------+------------------+
|summary|             price|
+-------+------------------+
|  count|             21613|
|   mean|       540088.1418|
| stddev|367127.19648270035|
|    min|             75000|
|    max|           7700000|
+-------+------------------+
{'fraction_non_alphanumeric': 0.15511904761904763, 'fraction_numerical': 0.08857142857142856, 'mean_word_length': 3.5337290879654613, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 18, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '343257', 'license': 'Apache-2.0', 'n_tokens_mistral': 3326, 'n_tokens_neox': 2872, 'n_words': 637, 'path': 'BigData-Leccion5.ipynb', 'repo_name': 'stevenjimbus/ProyectoBigData', 'size': '8411'}