<jupyter_start><jupyter_text>DATASETThe dataset is taken from **Kaggle** and the link is shared below. The title is 'Brooklyn Home Sales, 2003 to 2017'.Size of the dataset - 391k * 111link - https://www.kaggle.com/tianhwu/brooklynhomes2003to2017  MEMBERSMax <NAME>  REQUIREMENTS ( Packages with versions )**Pyspark is used to execute this notebook** - It is integration of python with spark framework.To install package - pip install 'package name'Spark - 2.3.1Python - Version > 3Numpy - 1.15.4Pandas - 0.23.4Matplotlib - 3.0.2Seaborn - 0.9.0  INSTALLATIONNo need to install any other packages apart from the ones listed below.Just make sure that you have python version > 3 installed on your machine.Run the spark session from the code cell below  FITTED MODELS**Regression Model** - Use this line - from pyspark.ml import feature, regression, evaluation, PipelineFour regression models are fitted with different set of input features to create the best modelRMSE is used as a parameter to get the best model  IMPORTING PACKAGES<jupyter_code>from pyspark.ml import feature
from pyspark.ml import clustering
from pyspark.ml import Pipeline
from pyspark.sql import functions as fn
import numpy as np
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.ml import feature, regression, evaluation, Pipeline
from pyspark.sql import functions as fn, Row
from pyspark.sql.functions import *
from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import OneHotEncoder
from pyspark.sql.types import StringType
from pyspark.sql.types import IntegerType
import matplotlib.pyplot as plt
import seaborn as sns; sns.set()<jupyter_output><empty_output><jupyter_text>INITIATE SPARK SESSION<jupyter_code>spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext<jupyter_output><empty_output><jupyter_text>**Load the dataset and print the schema**<jupyter_code>df = spark.read.option('header','true').option('inferSchema','true').csv('brooklyn_sales_map.csv')
print(df.printSchema())<jupyter_output>root
 |-- _c0: integer (nullable = true)
 |-- borough1: integer (nullable = true)
 |-- neighborhood: string (nullable = true)
 |-- building_class_category: string (nullable = true)
 |-- tax_class: string (nullable = true)
 |-- block: integer (nullable = true)
 |-- lot: integer (nullable = true)
 |-- easement: string (nullable = true)
 |-- building_class: string (nullable = true)
 |-- address9: string (nullable = true)
 |-- apartment_number: string (nullable = true)
 |-- zip_code: integer (nullable = true)
 |-- residential_units: integer (nullable = true)
 |-- commercial_units: integer (nullable = true)
 |-- total_units: integer (nullable = true)
 |-- land_sqft: decimal(10,0) (nullable = true)
 |-- gross_sqft: decimal(10,0) (nullable = true)
 |-- year_built: integer (nullable = true)
 |-- tax_class_at_sale: integer (nullable = true)
 |-- building_class_at_sale: string (nullable = true)
 |-- sale_price: decimal(10,0) (nullable = true)
 |-- sale_date: timestamp (nullable = true)
 |-- year[...]<jupyter_text>DATA EXPLORATION AND DESCRIPTIVE STATISTICS<jupyter_code>df = df.where(fn.col('sale_price')> 50000)  # Selecting rows where the sale price is greater than 50000
df = df.where(fn.col('sale_price') <= 14425000) # Selecting rows where the sale price is less than= 14425000
df = df.where(fn.col('neighborhood') != '3004')# Removing outlier
sale_price = df.select('sale_price').toPandas()
neighborhoods = df.select('neighborhood').toPandas()
sale_price = df.select('sale_price').toPandas()

sale_price = sale_price.astype('float')
sale_price.plot(kind='hist', bins=40)
plt.title('Sale Price Distribution')
plt.show()
sqft_sale = df.select('sale_price', 'gross_sqft').toPandas()
sqft_sale['sale_price'] = sqft_sale['sale_price'].astype('float')
sqft_sale['gross_sqft'] = sqft_sale['gross_sqft'].astype('float')

#take out positive outliers
sqft_sale = sqft_sale[sqft_sale['gross_sqft'] < 100000]

sqft_sale.plot(x = 'sale_price', y= 'gross_sqft', kind = 'scatter')
plt.title('Sale Price vs. Gross Sq. Ft. Area')
plt.show()
neighborhood_sale = df.select('neighborhood', 'sale_price').toPandas()
neighborhood_sale['sale_price'] = neighborhood_sale['sale_price'].astype('int')
neighborhood_sale = neighborhood_sale[neighborhood_sale['neighborhood'] != 'JAMAICA BAY']

ns = neighborhood_sale.groupby(by = 'neighborhood').mean()


fig = plt.figure(figsize=(20, 2))
ax = fig.add_subplot(111)
ax.bar(ns.index, ns['sale_price'])
plt.xticks(rotation= 90)
plt.title('Average Home Sale Price by Neighborhood')
import pylab
pylab.savefig('brooklyn_bar.png')
plt.show()<jupyter_output><empty_output><jupyter_text>DATA CLEANING**Removing NA's****Casting columns to required data type**<jupyter_code>df = df.withColumn("tax_class_at_sale", df["tax_class_at_sale"].cast(StringType()))
cols = [ 'RetailArea','StrgeArea','ProxCode','ExemptLand','borough1','zip_code', 'total_units',
                     'ExemptTot','YearAlter1','BuiltFar','block', 'lot', 'residential_units','commercial_units',
              'sale_price', 'gross_sqft', 'land_sqft', 'year_built', 'year_of_sale','building_class','tax_class_at_sale']

df = df.na.replace('NA','0')
df = df.replace('None','0')
df = df.dropna(subset=cols)<jupyter_output><empty_output><jupyter_text>ONE HOT ENCODING**Converting string features to numeric and then applying one hot encoding**<jupyter_code>indexer = StringIndexer(inputCol="neighborhood", outputCol="neighborhood_numeric").fit(df)
indexed_df = indexer.transform(df)


indexer1 = StringIndexer(inputCol="HistDist", outputCol="HistDist_numeric").fit(df)
indexed_df1 = indexer1.transform(df)

indexer2 = StringIndexer(inputCol="building_class", outputCol="building_class_numeric").fit(df)
indexed_df2 = indexer2.transform(df)


indexer3 = StringIndexer(inputCol="tax_class_at_sale", outputCol="tax_class_at_sale_numeric").fit(df)
indexed_df3 = indexer3.transform(df)

# One hot encoding

encoder = OneHotEncoder(inputCol="neighborhood_numeric", outputCol="neighborhood_vector")
encoded_df = encoder.transform(indexed_df)


encoder1 = OneHotEncoder(inputCol="HistDist_numeric", outputCol="HistDist_vector")
encoded_df1 = encoder1.transform(indexed_df1)

encoder2 = OneHotEncoder(inputCol="building_class_numeric", outputCol="building_class_vector")
encoded_df2 = encoder2.transform(indexed_df2)

encoder3 = OneHotEncoder(inputCol="tax_class_at_sale_numeric", outputCol="tax_class_at_sale_vector")
encoded_df3 = encoder3.transform(indexed_df3)

encoder_df = encoded_df.select('neighborhood_vector')
encoder_df1 = encoded_df1.select('HistDist_vector')
encoder_df2 = encoded_df2.select('building_class_vector')
encoder_df3 = encoded_df3.select('tax_class_at_sale_vector')<jupyter_output><empty_output><jupyter_text>**Using SQL join to combine different dataframes created above to one final dataframe****Using monotonically_increasing_id() to apply JOIN**<jupyter_code>from pyspark.sql import functions as fn
df = df.withColumn("idx", fn.monotonically_increasing_id())

encoder_df=encoder_df.withColumn("idx", fn.monotonically_increasing_id())

encoder_df1=encoder_df1.withColumn("idx", fn.monotonically_increasing_id())

encoder_df2=encoder_df2.withColumn("idx", fn.monotonically_increasing_id())

encoder_df3=encoder_df3.withColumn("idx", fn.monotonically_increasing_id())

df = df.join(encoder_df, df.idx ==encoder_df.idx )
df = df.drop('idx')

df = df.withColumn("idx", fn.monotonically_increasing_id())
df = df.join(encoder_df1, df.idx ==encoder_df1.idx )
df = df.drop('idx')

df = df.withColumn("idx", fn.monotonically_increasing_id())
df = df.join(encoder_df2, df.idx ==encoder_df2.idx )
df = df.drop('idx')

df = df.withColumn("idx", fn.monotonically_increasing_id())
df = df.join(encoder_df3, df.idx ==encoder_df3.idx )
df = df.drop('idx')<jupyter_output><empty_output><jupyter_text>TYPE CASTING**Converting following columns to Integer Type - So that we can apply scaling and then create a model****Printing Schema to verify the cast result**<jupyter_code>df = df.withColumn("RetailArea", df["RetailArea"].cast(IntegerType()))
df = df.withColumn("StrgeArea", df["StrgeArea"].cast(IntegerType()))
df = df.withColumn("ProxCode", df["ProxCode"].cast(IntegerType()))

df = df.withColumn("ExemptLand", df["ExemptLand"].cast(IntegerType()))
df = df.withColumn("ExemptTot", df["ExemptTot"].cast(IntegerType()))
df = df.withColumn("YearAlter1", df["YearAlter1"].cast(IntegerType()))
df = df.withColumn("BuiltFar", df["BuiltFar"].cast(IntegerType()))

df = df.withColumn("gross_sqft", df["gross_sqft"].cast(IntegerType()))
df = df.withColumn("land_sqft", df["land_sqft"].cast(IntegerType()))
df = df.withColumn("year_built", df["year_built"].cast(IntegerType()))
df = df.withColumn("year_of_sale", df["year_of_sale"].cast(IntegerType()))
print(df.printSchema())<jupyter_output>root
 |-- _c0: integer (nullable = true)
 |-- borough1: integer (nullable = true)
 |-- neighborhood: string (nullable = true)
 |-- building_class_category: string (nullable = true)
 |-- tax_class: string (nullable = true)
 |-- block: integer (nullable = true)
 |-- lot: integer (nullable = true)
 |-- easement: string (nullable = true)
 |-- building_class: string (nullable = true)
 |-- address9: string (nullable = true)
 |-- apartment_number: string (nullable = true)
 |-- zip_code: integer (nullable = true)
 |-- residential_units: integer (nullable = true)
 |-- commercial_units: integer (nullable = true)
 |-- total_units: integer (nullable = true)
 |-- land_sqft: integer (nullable = true)
 |-- gross_sqft: integer (nullable = true)
 |-- year_built: integer (nullable = true)
 |-- tax_class_at_sale: string (nullable = true)
 |-- building_class_at_sale: string (nullable = true)
 |-- sale_price: decimal(10,0) (nullable = true)
 |-- sale_date: timestamp (nullable = true)
 |-- year_of_sale: int[...]<jupyter_text>SAMPLING - Creating Test Splits**Splitting the dataset into 3 parts training(60%), validation(30%) and testing(10%)**<jupyter_code>training_df, validation_df, testing_df = df.randomSplit([0.6, 0.3, 0.1], seed=0)<jupyter_output><empty_output><jupyter_text>CREATING PIPELINES**stage1- Vector assembling****stage2- Standard scaling**<jupyter_code>Pipeline1 = Pipeline(stages=[
    feature.VectorAssembler(inputCols=[ 'gross_sqft', 'land_sqft', 'year_built', 'year_of_sale','block', 'lot', 'total_units','residential_units','commercial_units'],
                           outputCol='features'),
    
    feature.StandardScaler(withStd=False,
                           inputCol='features', outputCol='zfeatures1')]).fit(df)
Pipeline2 = Pipeline(stages=[
    feature.VectorAssembler(inputCols=[ 'neighborhood_vector','gross_sqft', 'land_sqft', 'year_built', 'year_of_sale','block', 'lot', 'total_units','residential_units','commercial_units'],
                           outputCol='features'),
    
    feature.StandardScaler(withStd=False,
                           inputCol='features', outputCol='zfeatures2')]).fit(df)
Pipeline3 = Pipeline(stages=[
    feature.VectorAssembler(inputCols=['block', 'lot', 'residential_units','commercial_units', 'RetailArea',
            'StrgeArea','ProxCode','ExemptLand','ExemptTot','YearAlter1','BuiltFar','neighborhood_vector','HistDist_vector'],
              outputCol='features'),
    feature.StandardScaler(withStd=False,
                           inputCol='features', outputCol='zfeatures3')]).fit(df)
Pipeline4 = Pipeline(stages=[
    feature.VectorAssembler(inputCols=['RetailArea','StrgeArea','ProxCode','ExemptLand',
                     'ExemptTot','YearAlter1','BuiltFar','block', 'lot', 'residential_units','commercial_units',
               'gross_sqft', 'land_sqft', 'year_built', 'year_of_sale',
              'neighborhood_vector','HistDist_vector','building_class_vector','tax_class_at_sale_vector'],
                             outputCol='features'),
    
    feature.StandardScaler(withStd=False,
                           inputCol='features', outputCol='zfeatures4')]).fit(df)<jupyter_output><empty_output><jupyter_text>FITTING MODELS**Regression Models****Features - generated above by the 4 pipelines****Label - sale_price**<jupyter_code>lr1= regression.LinearRegression(featuresCol='zfeatures1', labelCol='sale_price')
lr2= regression.LinearRegression(featuresCol='zfeatures2', labelCol='sale_price')
lr3= regression.LinearRegression(featuresCol='zfeatures3', labelCol='sale_price')
lr4= regression.LinearRegression(featuresCol='zfeatures4', labelCol='sale_price')
pipeline_lr1 = Pipeline(stages=[Pipeline1,lr1])

model1 = pipeline_lr1.fit(training_df)
pipeline_lr2 = Pipeline(stages=[Pipeline2,lr2])

model2 = pipeline_lr2.fit(training_df)
pipeline_lr3 = Pipeline(stages=[Pipeline3,lr3])

model3 = pipeline_lr3.fit(training_df)
pipeline_lr4 = Pipeline(stages=[Pipeline4,lr4])

model4 = pipeline_lr4.fit(training_df)<jupyter_output><empty_output><jupyter_text>GENERALISATION**transforming the fitted models on validation set and then calculating the lowest model RMSE on the test set**<jupyter_code># Make predictions.
predictions1 = model1.transform(validation_df)

predictions2 = model2.transform(validation_df)

predictions3 = model3.transform(validation_df)

predictions4 = model4.transform(validation_df)<jupyter_output><empty_output><jupyter_text>CALCULATE RMSE - VALIDATION SET<jupyter_code>mse = fn.mean((fn.col('sale_price') - fn.col('prediction'))**2).alias('mse')
rmse = sqrt(mse)

print('Model 1 - RMSE')
model1.transform(validation_df).select(rmse).show()

print('Model 2 - RMSE')
model2.transform(validation_df).select(rmse).show()

print('Model 3 - RMSE')
model3.transform(validation_df).select(rmse).show()

print('Model 4 - RMSE')
model4.transform(validation_df).select(rmse).show()<jupyter_output>Model 1 - RMSE
+-------------------------------------------------------+
|SQRT(avg(POWER((sale_price - prediction), 2)) AS `mse`)|
+-------------------------------------------------------+
|                                      841982.7225376875|
+-------------------------------------------------------+

Model 2 - RMSE
+-------------------------------------------------------+
|SQRT(avg(POWER((sale_price - prediction), 2)) AS `mse`)|
+-------------------------------------------------------+
|                                       821859.499182439|
+-------------------------------------------------------+

Model 3 - RMSE
+-------------------------------------------------------+
|SQRT(avg(POWER((sale_price - prediction), 2)) AS `mse`)|
+-------------------------------------------------------+
|                                      871714.8577946433|
+-------------------------------------------------------+

Model 4 - RMSE
+-------------------------------------------------------+
|SQRT(avg[...]<jupyter_text>CALCULATE RMSE for the best model so far - TEST SET<jupyter_code>model4.transform(testing_df).select(rmse).show() # best model<jupyter_output>+-------------------------------------------------------+
|SQRT(avg(POWER((sale_price - prediction), 2)) AS `mse`)|
+-------------------------------------------------------+
|                                      721667.1557282272|
+-------------------------------------------------------+<jupyter_text>PRINT COEFFICIENTS AND INTERCEPT FOR THE BEST MODEL<jupyter_code>print(model4.stages[-1].coefficients)
print(model4.stages[-1].intercept)<jupyter_output>[53.39489829640317,0.3359297617966068,-15439.597234871168,-0.07930629884127874,0.016182086701890946,34.13915230025102,13634.41869129464,17.669645827888882,90.21048644925759,169.50132446483005,38592.714126976345,8.03006542712752,16.38234689576356,8.489952586101458,41122.74602876816,-1875760.6805362294,-2085615.8180195158,-1685035.836424141,-1992398.353327278,-1740122.8343524777,-1165886.7981119999,-2073896.9981284307,-1838250.479005901,-1798516.989673861,-2026780.660221137,-1873888.6345046272,-1724253.5948887495,-1753452.8198705511,-1677695.6365681123,-2020097.6150297825,-1769550.6159479904,-1053788.7031492435,-1751590.42692453,-1706895.7774222533,-2064587.020432294,-1557292.478393868,-1247350.388043273,-1551083.8376124732,-1495115.3032635383,-1997746.4105519284,-1818230.079721739,-1322755.187666005,-1639331.130327439,-1969970.9359950523,-1786941.337432396,-1778890.2428351762,-1416928.1904097474,-1363109.4217749191,-2048989.2051643727,-2001072.4215702056,-2095237.959577065,-972019.41740[...]<jupyter_text>SHOW PREDICTIONS<jupyter_code># Select example rows to display.
pred1 = predictions1.select("zfeatures1","sale_price","prediction")
pred2 = predictions2.select("zfeatures2","sale_price","prediction")
pred3 = predictions3.select("zfeatures3","sale_price","prediction")
pred4 = predictions4.select("zfeatures4","sale_price","prediction")<jupyter_output><empty_output><jupyter_text>**Coverting predicted dataframes to pandas dataframes**<jupyter_code>df_pred1 = pred1.toPandas()
df_pred2 = pred2.toPandas()
df_pred3 = pred3.toPandas()
df_pred4 = pred4.toPandas()<jupyter_output><empty_output><jupyter_text>PLOTTING PREDICTIONS - Actual vs Predicted**scatter plot****red dots - predicted, blue dots - actual sale price****scale - xaxis = predicted sale price, yaxis = actual sale price**  MODEL 1<jupyter_code>pred1.show(5)
Ypred = df_pred1['prediction']
Ytest = df_pred1['sale_price']
colors=('red','blue')

plt.figure(figsize=(10,6))
plt.scatter(Ypred,Ytest,c=colors)

plt.xlabel('sale_price_pred')
plt.ylabel('sale_price_test')
plt.show<jupyter_output><empty_output><jupyter_text>MODEL 2<jupyter_code>pred2.show(5)
Ypred = df_pred2['prediction']
Ytest = df_pred2['sale_price']
colors=('red','blue')

plt.figure(figsize=(10,6))
plt.scatter(Ypred,Ytest,c=colors)

#xlim(1, 50)

plt.xlabel('sale_price_pred')
plt.ylabel('sale_price_test')
plt.show<jupyter_output><empty_output><jupyter_text>MODEL 3<jupyter_code>pred3.show(5)
Ypred = df_pred3['prediction']
Ytest = df_pred3['sale_price']
colors=('red','blue')

plt.figure(figsize=(10,6))
plt.scatter(Ypred,Ytest,c=colors)

#xlim(1, 50)

plt.xlabel('sale_price_pred')
plt.ylabel('sale_price_test')
plt.show<jupyter_output><empty_output><jupyter_text>MODEL 4 - Best Model<jupyter_code>pred4.show(5)
Ypred = df_pred4['prediction']
Ytest = df_pred4['sale_price']
colors=('red','blue')

plt.figure(figsize=(10,6))
plt.scatter(Ypred,Ytest,c=colors)

#xlim(1, 50)

plt.xlabel('sale_price_pred')
plt.ylabel('sale_price_test')
plt.show<jupyter_output><empty_output><jupyter_text>PLOTTING RMSE for all the models<jupyter_code>import pandas as pd
df_rmse = pd.DataFrame({'Linear Regression Models':['LR_Model1', 'LR_Model2', 'LR_Model3','LR_Model4'], 'RMSE':[841982.7225376875,821859.499182439,871714.8577946433,732417.5763206462]})
ax = df_rmse.plot.bar(x='Linear Regression Models', y='RMSE', rot=0,title='Linear Regression Models vs RMSE')<jupyter_output><empty_output><jupyter_text>DEMO **Enter your own values to preict the sale price - Steps below shows the predicted price for model 1**<jupyter_code>sample = [Row(gross_sqft = 3465000, land_sqft= 3000, year_built = 2000,
              year_of_sale= 2014, block = 120, lot =1, total_units = 4, residential_units=4, commercial_units=0,sale_price = '')]
sample_df=spark.createDataFrame(sample)
predictions5 = model1.transform(sample_df)
predictions5.select("zfeatures1","sale_price","prediction")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17667532467532468, 'fraction_numerical': 0.06831168831168831, 'mean_word_length': 5.004678727386151, 'pattern_counts': {'":': 0, '<': 91, '<?xml version=': 0, '>': 92, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '623208', 'license': 'MIT', 'n_tokens_mistral': 7344, 'n_tokens_neox': 6446, 'n_words': 1091, 'path': 'Project_718_FinalCode.ipynb', 'repo_name': 'ManishDalal/Data-Science', 'size': '19277'}