<jupyter_start><jupyter_text>Predicting CA housing prices using SparkMLib Table of ContentsBoiler plate - initialize SparkSession &amp; ContextAbout CA housing datasetPreprocess dataConvert RDD to Spark DataFrameExploratory data analysisFeature engineeringRe-order columns and split table into label and featuresScale data by shifting mean to 0 and making SD = 1Split data into training and test setsPerform Multiple RegressionInspect model propertiesPerform predictionsRegression evaluatorErrors - MAE, RMSECompare training vs prediction errorsExport data as a Pandas DataFrameWrite to disk as CSVPublish to GISSpark jobs  Boiler plate - initialize SparkSession & Context<jupyter_code># Import SparkSession
from pyspark.sql import SparkSession

# Build the SparkSession
spark = SparkSession.builder \
   .master("local") \
   .appName("Linear Regression Model") \
   .config("spark.executor.memory", "1gb") \
   .getOrCreate()
   
sc = spark.sparkContext<jupyter_output><empty_output><jupyter_text>About CA housing dataset Number of records: 20640variables: Lat, Long, Median Age, rooms, bedrooms, population in block, households, med income, med house value  Preprocess data<jupyter_code>!ls ../datasets/CaliforniaHousing/
# load data file
rdd = sc.textFile('../datasets/CaliforniaHousing/cal_housing.data')

# load header
header = sc.textFile('../datasets/CaliforniaHousing/cal_housing.domain')
len(rdd.collect())
len(rdd.take(5))
rdd.take(5)
# split by comma
rdd = rdd.map(lambda line : line.split(','))

# get the first two lines
rdd.first()<jupyter_output><empty_output><jupyter_text>Convert RDD to Spark DataFrame<jupyter_code># convert RDD to a dataframe
from pyspark.sql import Row

# Map the RDD to a DF
df = rdd.map(lambda line: Row(longitude=line[0], 
                              latitude=line[1], 
                              housingMedianAge=line[2],
                              totalRooms=line[3],
                              totalBedRooms=line[4],
                              population=line[5], 
                              households=line[6],
                              medianIncome=line[7],
                              medianHouseValue=line[8])).toDF()

# show the top few DF rows
df.show(5)
df.printSchema()
# convert all strings to float using a User Defined Function

from pyspark.sql.types import *

def cast_columns(df):
    for column in df.columns:
        df = df.withColumn(column, df[column].cast(FloatType()))
    return df

new_df = cast_columns(df)
new_df.show(2)
new_df.printSchema()<jupyter_output>root
 |-- households: float (nullable = true)
 |-- housingMedianAge: float (nullable = true)
 |-- latitude: float (nullable = true)
 |-- longitude: float (nullable = true)
 |-- medianHouseValue: float (nullable = true)
 |-- medianIncome: float (nullable = true)
 |-- population: float (nullable = true)
 |-- totalBedRooms: float (nullable = true)
 |-- totalRooms: float (nullable = true)<jupyter_text>Exploratory data analysisPrint the summary stats of the table<jupyter_code>new_df.describe().show()<jupyter_output>+-------+-----------------+------------------+-----------------+-------------------+------------------+------------------+------------------+-----------------+------------------+
|summary|       households|  housingMedianAge|         latitude|          longitude|  medianHouseValue|      medianIncome|        population|    totalBedRooms|        totalRooms|
+-------+-----------------+------------------+-----------------+-------------------+------------------+------------------+------------------+-----------------+------------------+
|  count|            20640|             20640|            20640|              20640|             20640|             20640|             20640|            20640|             20640|
|   mean|499.5396802325581|28.639486434108527|35.63186143109965|-119.56970444871473|206855.81690891474|3.8706710030346416|1425.4767441860465|537.8980135658915|2635.7630813953488|
| stddev|382.3297528316098| 12.58555761211163|2.135952380602968|  2.003531742932898|115395.61587441359|1.[...]<jupyter_text>Feature engineeringAdd more columns such as 'number of bedrooms per room', 'rooms per household'. Also scale the 'medianHouseValue' by 1000 so it falls within range of other numbers.<jupyter_code>from pyspark.sql.functions import col

df = df.withColumn('medianHouseValue', col('medianHouseValue')/100000)
df.first()
# add rooms per household
df = df.withColumn('roomsPerHousehold', col('totalRooms')/col('households'))

# add population per household (num people in the home)
df = df.withColumn('popPerHousehold', col('population')/col('households'))

# add bedrooms per room
df = df.withColumn('bedroomsPerRoom', col('totalBedRooms')/col('totalRooms'))
df.first()<jupyter_output><empty_output><jupyter_text>Re-order columns and split table into label and features<jupyter_code>df.columns
df = df.select('medianHouseValue','households',
 'housingMedianAge',
 'latitude',
 'longitude',
 'medianIncome',
 'population',
 'totalBedRooms',
 'totalRooms',
 'roomsPerHousehold',
 'popPerHousehold',
 'bedroomsPerRoom')<jupyter_output><empty_output><jupyter_text>Create a new DataFrame that explicitly labels the columns as labels and features. `DenseVector` is used to temporarily convert the data into numpy array and regroup into a named column DataFrame<jupyter_code>from pyspark.ml.linalg import DenseVector

# return a tuple of first column and all other columns
temp_data = df.rdd.map(lambda x:(x[0], DenseVector(x[1:])))

#construct back a new DataFrame
df2 = spark.createDataFrame(temp_data, ['label','features'])
df2.take(2)<jupyter_output><empty_output><jupyter_text>Scale data by shifting mean to 0 and making SD = 1This ensures all columns have similar levels of variability<jupyter_code># use StandardScaler to scale the features to std normal distribution
from pyspark.ml.feature import StandardScaler

s_scaler_model = StandardScaler(inputCol='features', outputCol='features_scaled')
scaler_fn = s_scaler_model.fit(df2)
scaled_df = scaler_fn.transform(df2)

scaled_df.take(2)<jupyter_output><empty_output><jupyter_text>Split data into training and test sets<jupyter_code>train_data, test_data = scaled_df.randomSplit([.8,.2], seed=101)
type(train_data)<jupyter_output><empty_output><jupyter_text>Perform Multiple RegressionTrain the model<jupyter_code>from pyspark.ml.regression import LinearRegression

lr = LinearRegression(labelCol='label', maxIter=20)

linear_model = lr.fit(train_data)<jupyter_output><empty_output><jupyter_text>Inspect model properties<jupyter_code>type(linear_model)
linear_model.coefficients<jupyter_output><empty_output><jupyter_text>Print columns and their coefficients<jupyter_code>list(zip(df.columns[1:], linear_model.coefficients))
linear_model.intercept
linear_model.summary.numInstances<jupyter_output><empty_output><jupyter_text>MAE from training data<jupyter_code>linear_model.summary.meanAbsoluteError * 100000<jupyter_output><empty_output><jupyter_text>Thus, MAE on training data is off by $50,000<jupyter_code>linear_model.summary.meanSquaredError
linear_model.summary.rootMeanSquaredError * 100000<jupyter_output><empty_output><jupyter_text>Thus, RMSE shows fitting on training data is off by $68,392<jupyter_code>list(zip(df.columns[1:], linear_model.summary.pValues))<jupyter_output><empty_output><jupyter_text>Perform predictions<jupyter_code>predicted = linear_model.transform(test_data)
predicted.columns
type(predicted)
test_predictions = predicted.select('prediction').rdd.map(lambda x:x[0])
test_labels = predicted.select('label').rdd.map(lambda x:x[0])

test_predictions_labels = test_predictions.zip(test_labels)
test_predictions_labels_df = spark.createDataFrame(test_predictions_labels, 
                                                   ['predictions','labels'])

test_predictions_labels_df.take(2)<jupyter_output><empty_output><jupyter_text>Regression evaluator<jupyter_code>from pyspark.ml.evaluation import RegressionEvaluator

linear_reg_eval = RegressionEvaluator(predictionCol='predictions', labelCol='labels')
linear_reg_eval.evaluate(test_predictions_labels_df)<jupyter_output><empty_output><jupyter_text>Errors - MAE, RMSE<jupyter_code># mean absolute error
prediction_mae = linear_reg_eval.evaluate(test_predictions_labels_df, 
                                          {linear_reg_eval.metricName:'mae'}) * 100000
prediction_mae
# RMSE
prediction_rmse = linear_reg_eval.evaluate(test_predictions_labels_df, 
                                           {linear_reg_eval.metricName:'rmse'}) * 100000

prediction_rmse<jupyter_output><empty_output><jupyter_text>Compare training vs prediction errors<jupyter_code>print('(training error, prediction error)')
print((linear_model.summary.rootMeanSquaredError * 100000, prediction_rmse))
print((linear_model.summary.meanAbsoluteError * 100000, prediction_mae))<jupyter_output>(training error, prediction error)
(68392.54514549255, 69622.95496358669)
(49805.60256405839, 49690.440586665725)<jupyter_text>Export data as a Pandas DataFrame<jupyter_code>predicted_pandas_df = predicted.select('features','prediction').toPandas()
predicted_pandas_df.head()
predicted_pandas_df.columns
import pandas as pd
predicted_pandas_df2 = pd.DataFrame(predicted_pandas_df['features'].values.tolist(), 
                                   columns=df.columns[1:])

predicted_pandas_df2.head()
predicted_pandas_df2['predictedHouseValue'] = predicted_pandas_df['prediction']<jupyter_output><empty_output><jupyter_text>Write to disk as CSV<jupyter_code>predicted_pandas_df2.to_csv('CA_house_prices_predicted.csv')
!ls
predicted_pandas_df2.shape<jupyter_output><empty_output><jupyter_text>Publish to GIS<jupyter_code>import pandas as pd
from arcgis.gis import GIS
gis = GIS("https://www.arcgis.com","arcgis_python")
from arcgis.features import SpatialDataFrame
sdf = SpatialDataFrame.from_csv('CA_house_prices_predicted.csv')
sdf.head(5)
houses_predicted_fc = gis.content.import_data(sdf[:999])
houses_predicted_fc
ca_map = gis.map('California')
ca_map<jupyter_output><empty_output><jupyter_text>![](./ca-house-price-map.png)<jupyter_code>ca_map.add_layer(houses_predicted_fc, {'renderer':'ClassedColorRenderer',
                                      'field_name':'predictedHouseValue'})<jupyter_output>
{'fraction_non_alphanumeric': 0.13329441525588637, 'fraction_numerical': 0.04388013232146332, 'mean_word_length': 4.2984536082474225, 'pattern_counts': {'":': 0, '<': 89, '<?xml version=': 0, '>': 89, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '120292', 'license': 'MIT', 'n_tokens_mistral': 3564, 'n_tokens_neox': 3195, 'n_words': 627, 'path': 'ml/pySpark/spark-ml-CA-housing.ipynb', 'repo_name': 'AtmaMani/pyChakras', 'size': '10278'}