<jupyter_start><jupyter_text>0. Dependencies & Setup  0.1. Related dependencies and environment variables<jupyter_code># Install spark-related dependencies
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://mirrors.viethosting.com/apache/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz
!tar xf spark-2.4.4-bin-hadoop2.7.tgz

!pip install -q findspark
!pip install pyspark

# Set up required environment variables
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.4.3-bin-hadoop2.7"<jupyter_output>Collecting pyspark
[?25l  Downloading https://files.pythonhosted.org/packages/87/21/f05c186f4ddb01d15d0ddc36ef4b7e3cedbeb6412274a41f26b55a650ee5/pyspark-2.4.4.tar.gz (215.7MB)
[K     |████████████████████████████████| 215.7MB 60kB/s 
[?25hCollecting py4j==0.10.7
[?25l  Downloading https://files.pythonhosted.org/packages/e3/53/c737818eb9a7dc32a7cd4f1396e787bd94200c3997c72c1dbe028587bd76/py4j-0.10.7-py2.py3-none-any.whl (197kB)
[K     |████████████████████████████████| 204kB 52.0MB/s 
[?25hBuilding wheels for collected packages: pyspark
  Building wheel for pyspark (setup.py) ... [?25l[?25hdone
  Created wheel for pyspark: filename=pyspark-2.4.4-py2.py3-none-any.whl size=216130387 sha256=f7c66a9814016bb50b8262a0b56bf81325d2757102820e109301cb7e1fbea4cd
  Stored in directory: /root/.cache/pip/wheels/ab/09/4d/0d184230058e654eb1b04467dbc1292f00eaa186544604b471
Successfully built pyspark
Installing collected packages: py4j, pyspark
Successfully installed py4j-0.10.7 pyspark-2.4.4<jupyter_text>0.2. Initialize Spark<jupyter_code>import findspark
findspark.init('spark-2.4.4-bin-hadoop2.7')
 
import pyspark
from pyspark import SparkConf
from pyspark.sql import SparkSession

spark = SparkSession \
.builder \
.appName("sparkml_lab") \
.master("local[*]") \
.getOrCreate()

# Check
spark.sparkContext.getConf().getAll()
spark<jupyter_output><empty_output><jupyter_text>1. Import Housing Dataset<jupyter_code>import tarfile
import urllib

DOWNLOAD_ROOT = "https://raw.githubusercontent.com/ageron/handson-ml2/master/"
HOUSING_PATH = os.path.join("datasets", "housing")
HOUSING_URL = DOWNLOAD_ROOT + "datasets/housing/housing.tgz"

def fetch_housing_data(housing_url=HOUSING_URL, housing_path=HOUSING_PATH):
    os.makedirs(housing_path, exist_ok=True)
    tgz_path = os.path.join(housing_path, "housing.tgz")
    urllib.request.urlretrieve(housing_url, tgz_path)
    housing_tgz = tarfile.open(tgz_path)
    housing_tgz.extractall(path=housing_path)
    housing_tgz.close()

fetch_housing_data()
import pandas as pd

def load_housing_data(housing_path=HOUSING_PATH):
    csv_path = os.path.join(housing_path, "housing.csv")
    return pd.read_csv(csv_path)

housing_pd = load_housing_data()<jupyter_output><empty_output><jupyter_text>2. Data Discovery<jupyter_code>#housing = spark.read.csv("/content/datasets/housing/housing.csv", header = True)
housing = spark.createDataFrame(housing_pd)<jupyter_output><empty_output><jupyter_text>2.1. Schema and dimensions Printing schema of the dataset<jupyter_code>housing.printSchema()<jupyter_output>root
 |-- longitude: double (nullable = true)
 |-- latitude: double (nullable = true)
 |-- housing_median_age: double (nullable = true)
 |-- total_rooms: double (nullable = true)
 |-- total_bedrooms: double (nullable = true)
 |-- population: double (nullable = true)
 |-- households: double (nullable = true)
 |-- median_income: double (nullable = true)
 |-- median_house_value: double (nullable = true)
 |-- ocean_proximity: string (nullable = true)<jupyter_text>Printing number of records in the dataset<jupyter_code>housing.count()<jupyter_output><empty_output><jupyter_text>2.2. Look at the data Printing first five records of the dataset<jupyter_code>housing.take(5)<jupyter_output><empty_output><jupyter_text>Printing the number of records with population more than 10000<jupyter_code>housing.where(housing.population > 10000).collect()<jupyter_output><empty_output><jupyter_text>2.3. Statistical summary Printing summary of the table statistics for the attributes housing_median_age, total_rooms, median_house_value, and population.<jupyter_code>housing.describe(["housing_median_age", "total_rooms", "median_house_value", "population"]).show()<jupyter_output>+-------+------------------+------------------+------------------+------------------+
|summary|housing_median_age|       total_rooms|median_house_value|        population|
+-------+------------------+------------------+------------------+------------------+
|  count|             20640|             20640|             20640|             20640|
|   mean|28.639486434108527|2635.7630813953488|206855.81690891474|1425.4767441860465|
| stddev|12.585557612111632|2181.6152515827957|115395.61587441375|1132.4621217653405|
|    min|               1.0|               2.0|           14999.0|               3.0|
|    max|              52.0|           39320.0|          500001.0|           35682.0|
+-------+------------------+------------------+------------------+------------------+<jupyter_text>Print the maximum age (housing_median_age), the minimum number of rooms (total_rooms), and the average of house values (median_house_value).<jupyter_code>value1 = housing.agg({"housing_median_age": "max"}).collect()[0]
value2 = housing.agg({"total_rooms" : "min"}).collect()[0]
value3 = housing.agg({"median_house_value" : "mean"}).collect()[0]

print("Maximum housing age: " + str(value1["max(housing_median_age)"]) + 
      "\nMinimum number of rooms: " + str(value2["min(total_rooms)"]) +
      "\nAverage house value: " + str(value3["avg(median_house_value)"]))<jupyter_output>Maximum housing age: 52.0
Minimum number of rooms: 2.0
Average house value: 206855.81690891474<jupyter_text>2.4. Data breakdown by categorical data Printing the number of houses in different areas (ocean_proximity), and sorted in descending order.<jupyter_code>housing.cube("ocean_proximity").count().sort('count', ascending=False).show()<jupyter_output>+---------------+-----+
|ocean_proximity|count|
+---------------+-----+
|           null|20640|
|      <1H OCEAN| 9136|
|         INLAND| 6551|
|     NEAR OCEAN| 2658|
|       NEAR BAY| 2290|
|         ISLAND|    5|
+---------------+-----+<jupyter_text>Printing the average value of the houses (median_house_value) in different areas (ocean_proximity); calling the new column avg_value when printing it.<jupyter_code>from pyspark.sql import functions as F

housing.\
select(["ocean_proximity", "median_house_value"]).\
groupBy("ocean_proximity").\
agg(F.mean("median_house_value").alias("avg_value")).\
show()<jupyter_output>+---------------+------------------+
|ocean_proximity|         avg_value|
+---------------+------------------+
|         ISLAND|          380440.0|
|     NEAR OCEAN|249433.97742663656|
|       NEAR BAY|259212.31179039303|
|      <1H OCEAN|240084.28546409807|
|         INLAND|124805.39200122119|
+---------------+------------------+<jupyter_text>Rewritten in SQL.<jupyter_code>housing.createOrReplaceTempView("housing")
spark.sql('''
          SELECT ocean_proximity, avg(median_house_value)
          AS avg_value
          FROM housing
          GROUP BY ocean_proximity
          ''').show()<jupyter_output>+---------------+------------------+
|ocean_proximity|         avg_value|
+---------------+------------------+
|         ISLAND|          380440.0|
|     NEAR OCEAN|249433.97742663656|
|       NEAR BAY|259212.31179039303|
|      <1H OCEAN|240084.28546409807|
|         INLAND|124805.39200122119|
+---------------+------------------+<jupyter_text>2.5. Correlation among attributes Printing the correlation among the attributes housing_median_age, total_rooms, median_house_value, and population. To do so, first we need to put these attributes into one vector. Then, compute the standard correlation coefficient (Pearson) between every pair of attributes in this new vector. To make a vector of these attributes, we use the VectorAssembler Transformer.<jupyter_code>from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(inputCols=["housing_median_age",
                                       "total_rooms",
                                       "median_house_value",
                                       "population"],
                             outputCol="attributes")

housing_attributes = assembler.transform(housing)

housing_attributes.show(5)
from pyspark.ml.stat import Correlation

corr = Correlation.corr(housing_attributes, "attributes")

corr.collect()[0]["pearson({})".format("attributes")].values<jupyter_output><empty_output><jupyter_text>2.6. Combine and make new attributes Now, let's try out various attribute combinations. In the given dataset, the total number of rooms in a block is not very useful, if we don't know how many households there are. What we really want is the number of rooms per household. Similarly, the total number of bedrooms by itself is not very useful, and we want to compare it to the number of rooms. And the population per household seems like also an interesting attribute combination to look at. To do so, we add the three new columns to the dataset as below. We will call the new dataset the housingExtra.```rooms_per_household = total_rooms / householdsbedrooms_per_room = total_bedrooms / total_roomspopulation_per_household = population / households```<jupyter_code>new_attributes1 = housing.withColumn("rooms_per_household",
                                    housing.total_rooms / housing.households)

new_attributes2 = new_attributes1.withColumn("bedrooms_per_room",
                                    new_attributes1.total_bedrooms / new_attributes1.total_rooms)

new_housing = new_attributes2.withColumn("population_per_household",
                                    new_attributes2.population / new_attributes2.households)

new_housing.select("rooms_per_household",
                   "bedrooms_per_room",
                   "population_per_household").show()<jupyter_output>+-------------------+-------------------+------------------------+
|rooms_per_household|  bedrooms_per_room|population_per_household|
+-------------------+-------------------+------------------------+
|  6.984126984126984|0.14659090909090908|      2.5555555555555554|
|  6.238137082601054|0.15579659106916466|       2.109841827768014|
|  8.288135593220339|0.12951601908657123|      2.8022598870056497|
| 5.8173515981735155|0.18445839874411302|       2.547945205479452|
|  6.281853281853282| 0.1720958819913952|      2.1814671814671813|
|  4.761658031088083|0.23177366702937977|       2.139896373056995|
| 4.9319066147859925|0.19289940828402366|      2.1284046692607004|
|  4.797527047913447|0.22132731958762886|      1.7882534775888717|
|  4.294117647058823| 0.2602739726027397|       2.026890756302521|
|  4.970588235294118| 0.1992110453648915|       2.172268907563025|
|  5.477611940298507|0.19709355131698456|       2.263681592039801|
|  4.772479564032698| 0.2146731373108764|      2.0490463215258[...]<jupyter_text>3. Data Preparation Before going through the Machine Learning steps, let's first rename the label column from median_house_value to label.<jupyter_code>housing_renamed = new_housing.withColumnRenamed("median_house_value", "label")<jupyter_output><empty_output><jupyter_text>Now, we want to separate the numerical attributes from the categorical attribute (ocean_proximity) and keep their column names in two different lists. Moreover, since we don't want to apply the same transformations to the predictors (features) and the label, we should remove the label attribute from the list of predictors.<jupyter_code>columns_features_num = housing_renamed.columns

columns_features_num.remove("ocean_proximity")
columns_features_num.remove("label")

columns_features_cat = ["ocean_proximity"]<jupyter_output><empty_output><jupyter_text>3.1. Prepare Continuous Attributes  Data cleaningMost Machine Learning algorithms cannot work with missing features, so we should take care of them. As a first step, let's find the columns with missing values in the numerical attributes. To do so, we can print the number of missing values of each continues attributes, listed in colNum.<jupyter_code>null_df = housing_renamed.select([F.count(F.when(F.isnan(i) | \
                                          F.col(i).contains('NA') | \
                                          F.col(i).contains('NULL') | \
                                          F.col(i).isNull(), i)).alias(i) \
                                  for i in housing_renamed.columns])

null_df.show()<jupyter_output>+---------+--------+------------------+-----------+--------------+----------+----------+-------------+-----+---------------+-------------------+-----------------+------------------------+
|longitude|latitude|housing_median_age|total_rooms|total_bedrooms|population|households|median_income|label|ocean_proximity|rooms_per_household|bedrooms_per_room|population_per_household|
+---------+--------+------------------+-----------+--------------+----------+----------+-------------+-----+---------------+-------------------+-----------------+------------------------+
|        0|       0|                 0|          0|           207|         0|         0|            0|    0|              0|                  0|              207|                       0|
+---------+--------+------------------+-----------+--------------+----------+----------+-------------+-----+---------------+-------------------+-----------------+------------------------+<jupyter_text>As we observerd above, the total_bedrooms and bedrooms_per_room attributes have some missing values. One way to take care of missing values is to use the Imputer Transformer, which completes missing values in a dataset, either using the mean or the median of the columns in which the missing values are located. To use it, you need to create an Imputer instance, specifying that you want to replace each attribute's missing values with the "median" of that attribute.<jupyter_code>from pyspark.ml.feature import Imputer

imputer = Imputer(inputCols = ["total_bedrooms", "bedrooms_per_room"],
                  outputCols = ["total_bedrooms", "bedrooms_per_room"])

housing_imputed = imputer.fit(housing_renamed).transform(housing_renamed)

housing_imputed.select(["total_bedrooms", "bedrooms_per_room"]).show(5)<jupyter_output>+--------------+-------------------+
|total_bedrooms|  bedrooms_per_room|
+--------------+-------------------+
|         129.0|0.14659090909090908|
|        1106.0|0.15579659106916466|
|         190.0|0.12951601908657123|
|         235.0|0.18445839874411302|
|         280.0| 0.1720958819913952|
+--------------+-------------------+
only showing top 5 rows<jupyter_text>ScalingOne of the most important transformations we need to apply to our data is feature scaling. With few exceptions, Machine Learning algorithms don't perform well when the input numerical attributes have very different scales. This is the case for the housing data: the total number of rooms ranges from about 6 to 39,320, while the median incomes only range from 0 to 15. Note that scaling the label attribues is generally not required.One way to get all attributes to have the same scale is to use standardization. In standardization, for each value, first it subtracts the mean value (so standardized values always have a zero mean), and then it divides by the variance so that the resulting distribution has unit variance. To do this, we can use the StandardScaler Estimator. To use StandardScaler, again we need to convert all the numerical attributes into a big vectore of features using VectorAssembler, and then call StandardScaler on that vector.<jupyter_code>vectas = VectorAssembler(inputCols = columns_features_num,
                         outputCol = "features")

housing_featured = vectas.transform(housing_imputed)
from pyspark.ml.feature import StandardScaler
scaler = StandardScaler(inputCol = "features",
                        outputCol = "scaled",
                        withStd = True)

scalerModel = scaler.fit(housing_featured)

housing_scaled = scalerModel.transform(housing_featured)
housing_scaled.printSchema()

housing_scaled.show(5)<jupyter_output>root
 |-- longitude: double (nullable = true)
 |-- latitude: double (nullable = true)
 |-- housing_median_age: double (nullable = true)
 |-- total_rooms: double (nullable = true)
 |-- total_bedrooms: double (nullable = true)
 |-- population: double (nullable = true)
 |-- households: double (nullable = true)
 |-- median_income: double (nullable = true)
 |-- label: double (nullable = true)
 |-- ocean_proximity: string (nullable = true)
 |-- rooms_per_household: double (nullable = true)
 |-- bedrooms_per_room: double (nullable = true)
 |-- population_per_household: double (nullable = true)
 |-- features: vector (nullable = true)
 |-- scaled: vector (nullable = true)

+---------+--------+------------------+-----------+--------------+----------+----------+-------------+--------+---------------+-------------------+-------------------+------------------------+--------------------+--------------------+
|longitude|latitude|housing_median_age|total_rooms|total_bedrooms|population|households|medi[...]<jupyter_text>3.2. Prepare categorical attributes After imputing and scaling the continuous attributes, we should take care of the categorical attributes. Let's first print the number of distict values of the categirical attribute ocean_proximity.<jupyter_code>from pyspark.sql.functions import countDistinct

housing_renamed.agg(countDistinct("ocean_proximity")).show()<jupyter_output>+-------------------------------+
|count(DISTINCT ocean_proximity)|
+-------------------------------+
|                              5|
+-------------------------------+<jupyter_text>String indexerMost Machine Learning algorithms prefer to work with numbers. So let's convert the categorical attribute ocean_proximity to numbers. To do so, we can use the StringIndexer that encodes a string column of labels to a column of label indices. The indices are in [0, numLabels), ordered by label frequencies, so the most frequent label gets index 0.<jupyter_code>from pyspark.ml.feature import StringIndexer

indexer = StringIndexer(inputCol = "ocean_proximity", 
                        outputCol = "ocean_proximity_indexed")

housing_indexed = indexer.fit(housing_renamed).transform(housing_renamed)

housing_indexed.show(10)
housing_indexed.agg(countDistinct("ocean_proximity_indexed")).show()<jupyter_output>+---------+--------+------------------+-----------+--------------+----------+----------+-------------+--------+---------------+-------------------+-------------------+------------------------+-----------------------+
|longitude|latitude|housing_median_age|total_rooms|total_bedrooms|population|households|median_income|   label|ocean_proximity|rooms_per_household|  bedrooms_per_room|population_per_household|ocean_proximity_indexed|
+---------+--------+------------------+-----------+--------------+----------+----------+-------------+--------+---------------+-------------------+-------------------+------------------------+-----------------------+
|  -122.23|   37.88|              41.0|      880.0|         129.0|     322.0|     126.0|       8.3252|452600.0|       NEAR BAY|  6.984126984126984|0.14659090909090908|      2.5555555555555554|                    3.0|
|  -122.22|   37.86|              21.0|     7099.0|        1106.0|    2401.0|    1138.0|       8.3014|358500.0|       NEAR BAY|  6.2[...]<jupyter_text>One-hot encodingNow we convert the label indices built in the last step into one-hot vectors. To do this, we can take advantage of the OneHotEncoderEstimator Estimator.<jupyter_code>from pyspark.ml.feature import OneHotEncoderEstimator

encoder = OneHotEncoderEstimator(inputCols = ["ocean_proximity_indexed"],
                                 outputCols = ["ocean_proximity_hot"])

housing_encoded = encoder.fit(housing_indexed).transform(housing_indexed)

housing_encoded.show()<jupyter_output>+---------+--------+------------------+-----------+--------------+----------+----------+-------------+--------+---------------+-------------------+-------------------+------------------------+-----------------------+-------------------+
|longitude|latitude|housing_median_age|total_rooms|total_bedrooms|population|households|median_income|   label|ocean_proximity|rooms_per_household|  bedrooms_per_room|population_per_household|ocean_proximity_indexed|ocean_proximity_hot|
+---------+--------+------------------+-----------+--------------+----------+----------+-------------+--------+---------------+-------------------+-------------------+------------------------+-----------------------+-------------------+
|  -122.23|   37.88|              41.0|      880.0|         129.0|     322.0|     126.0|       8.3252|452600.0|       NEAR BAY|  6.984126984126984|0.14659090909090908|      2.5555555555555554|                    3.0|      (4,[3],[1.0])|
|  -122.22|   37.86|              21.0|     7099.0| [...]<jupyter_text>4. Pipeline As you can see, there are many data transformation steps that need to be executed in the right order. For example, we called the Imputer, VectorAssembler, and StandardScaler from left to right. However, we can use the Pipeline class to define a sequence of Transformers/Estimators, and run them in order. A Pipeline is an Estimator, thus, after a Pipeline's fit() method runs, it produces a PipelineModel, which is a Transformer.Now, let's create a pipeline called numPipeline to call the numerical transformers you built above (imputer, va, and scaler) in the right order from left to right, as well as a pipeline called catPipeline to call the categorical transformers (indexer and encoder). Then, put these two pipelines numPipeline and catPipeline into one pipeline.<jupyter_code>from pyspark.ml import Pipeline, PipelineModel

# Imputer
imp = Imputer(inputCols = ["total_bedrooms", "bedrooms_per_room"],
              outputCols = ["total_bedrooms", "bedrooms_per_room"])

# VectorAssembler
vec = VectorAssembler(inputCols = columns_features_num,
                      outputCol = "vfeatures")

# StandardScaler
sca = StandardScaler(inputCol = "vfeatures",
                     outputCol = "scaled",
                     withStd = True)

# numPipeline
numPipeline = Pipeline(stages = [imp, vec, sca])
# Indexer
ind = StringIndexer(inputCol = "ocean_proximity", 
                    outputCol = "ocean_proximity_indexed")

# Encoder
enc = OneHotEncoderEstimator(inputCols = ["ocean_proximity_indexed"],
                             outputCols = ["ocean_proximity_hot"])

# catPipeline
catPipeline = Pipeline(stages = [ind, enc])
# Pipeline Assembly
pipeline = Pipeline(stages = [numPipeline, catPipeline])

# Fit and transform
housing_new = pipeline.fit(housing_renamed).transform(housing_renamed)

housing_new.show(5)<jupyter_output>+---------+--------+------------------+-----------+--------------+----------+----------+-------------+--------+---------------+-------------------+-------------------+------------------------+--------------------+--------------------+-----------------------+-------------------+
|longitude|latitude|housing_median_age|total_rooms|total_bedrooms|population|households|median_income|   label|ocean_proximity|rooms_per_household|  bedrooms_per_room|population_per_household|           vfeatures|              scaled|ocean_proximity_indexed|ocean_proximity_hot|
+---------+--------+------------------+-----------+--------------+----------+----------+-------------+--------+---------------+-------------------+-------------------+------------------------+--------------------+--------------------+-----------------------+-------------------+
|  -122.23|   37.88|              41.0|      880.0|         129.0|     322.0|     126.0|       8.3252|452600.0|       NEAR BAY|  6.984126984126984|0.14659090909090[...]<jupyter_text>Now, we use VectorAssembler to put all attributes of the final dataset housing_new into a big vector, and call the new column features.<jupyter_code>newVec = VectorAssembler(inputCols = ["scaled", "ocean_proximity_hot"],
                         outputCol = "features")

dataset = newVec.transform(housing_new).select("features", "label")

dataset.show(5)<jupyter_output>+--------------------+--------+
|            features|   label|
+--------------------+--------+
|[-61.007269596069...|452600.0|
|[-61.002278409814...|358500.0|
|[-61.012260782324...|352100.0|
|[-61.017251968579...|341300.0|
|[-61.017251968579...|342200.0|
+--------------------+--------+
only showing top 5 rows<jupyter_text>5. Make a model Here we going to make four different regression models:*   Linear regression model*   Decision tree regression*   Random forest regression*   Gradient-Boosted Forest RegressionBut, before giving the data to train a Machine Learning model, let's first split the data into training dataset (trainSet) with 80% of the whole data, and test dataset (testSet) with 20% of it.<jupyter_code>training_set, test_set = dataset.randomSplit([0.8, 0.2], 1)<jupyter_output><empty_output><jupyter_text>5.1. Linear regression model Now, we train a Linear Regression model using the LinearRegression class. Then, print the coefficients and intercept of the model, as well as the summary of the model over the training set by calling the summary method.<jupyter_code>from pyspark.ml.regression import LinearRegression

lr = LinearRegression()

lrModel = lr.fit(training_set)
print(lrModel.coefficients)
print(lrModel.summary.rootMeanSquaredError)<jupyter_output>[-54477.3715845959,-55101.84003302382,13288.061745034573,10009.769723216179,710.726908834572,-52958.800287261125,46244.11745493835,77314.65490018335,5716.086889683943,15958.634084931,1399.7525958185972,-175517.1241143946,-211742.57639440673,-171874.40244509815,-180607.27387195334]
67516.54425720422<jupyter_text>Now, we use RegressionEvaluator to measure the root-mean-square-erroe (RMSE) of the model on the test dataset.<jupyter_code>from pyspark.ml.evaluation import RegressionEvaluator

predictions = lrModel.transform(test_set)
predictions.select("prediction", "label", "features").show(5)

evaluator = RegressionEvaluator()
rmse = evaluator.evaluate(predictions)
print(rmse)<jupyter_output>+------------------+--------+--------------------+
|        prediction|   label|            features|
+------------------+--------+--------------------+
|182697.27426641854|111400.0|[-62.020480405854...|
|146301.48449560022| 58100.0|[-61.995524474579...|
|156240.51035895618| 72200.0|[-61.980550915813...|
| 162194.0578350625| 70200.0|[-61.980550915813...|
|199432.91476045595|128900.0|[-61.975559729558...|
+------------------+--------+--------------------+
only showing top 5 rows

69613.0558778766<jupyter_text>5.2. Decision tree regression Repeat what you have done on Regression Model to build a Decision Tree model. We use the DecisionTreeRegressor to make a model and then measure its RMSE on the test dataset.<jupyter_code>from pyspark.ml.regression import DecisionTreeRegressor

dtr = DecisionTreeRegressor()

dtrModel = dtr.fit(training_set)

predictions = dtrModel.transform(test_set)

rmse = evaluator.evaluate(predictions)
print(rmse)<jupyter_output>68322.33909628939<jupyter_text>5.3. Random forest regression Let's try the test error on a Random Forest Model. We can use the RandomForestRegressor to make a Random Forest model.<jupyter_code>from pyspark.ml.regression import RandomForestRegressor

rfr = RandomForestRegressor()

rfrModel = rfr.fit(training_set)

predictions = rfrModel.transform(test_set)

rmse = evaluator.evaluate(predictions)
print(rmse)<jupyter_output>65335.06730164033<jupyter_text>5.4. Gradient Boosted Tree Regression Fianlly, we want to build a Gradient-boosted Tree Regression model and test the RMSE of the test data. We use the GBTRegressor to build the model.<jupyter_code>from pyspark.ml.regression import GBTRegressor

gbtr = GBTRegressor()

gbtrModel = gbtr.fit(training_set)

predictions = gbtrModel.transform(test_set)

rmse = evaluator.evaluate(predictions)
print(rmse)<jupyter_output>56418.348519374646<jupyter_text>6. Hyperparameter Tuning An important task in Machie Learning is model selection, or using data to find the best model or parameters for a given task. This is also called tuning. Tuning may be done for individual Estimators such as LinearRegression, or for entire Pipelines which include multiple algorithms, featurization, and other steps. Users can tune an entire Pipeline at once, rather than tuning each element in the Pipeline separately. MLlib supports model selection tools, such as CrossValidator. These tools require the following items:*   Estimator*   ParamMaps*   EvaluatorCrossValidator begins by splitting the dataset into a set of folds, which are used as separate training and test datasets. For example with k=3 folds, CrossValidator will generate 3 (training, test) dataset pairs, each of which uses 2/3 of the data for training and 1/3 for testing. To evaluate a particular ParamMap, CrossValidator computes the average evaluation metric for the 3 Models produced by fitting the Estimator on the 3 different (training, test) dataset pairs. After identifying the best ParamMap, CrossValidator finally re-fits the Estimator using the best ParamMap and the entire dataset.Below, we use the CrossValidator to select the best Random Forest model. To do so, we need to define a grid of parameters. Let's say we want to do the search among the different number of trees (1, 5, and 10), and different tree depth (5, 10, and 15).<jupyter_code>from pyspark.ml.tuning import ParamGridBuilder, CrossValidator

paramGrid = ParamGridBuilder().\
            addGrid(rfr.numTrees, [1, 5, 10]).\
            addGrid(rfr.maxDepth, [5, 10, 15]).build()

cv = CrossValidator(estimator = rfr,
                    estimatorParamMaps = paramGrid,
                    evaluator = evaluator,
                    numFolds = 3)

cvModel = cv.fit(training_set)

predictions = cvModel.transform(test_set)

rmse = evaluator.evaluate(predictions)

print(rmse)<jupyter_output>53068.69884031164<jupyter_text>7. Custom transformer At the end of part two, we added extra columns to the housing dataset. Here, we are going to implement a Transformer to do the same task. The Transformer should take the name of two input columns inputCol1 and inputCol2, as well as the name of ouput column outputCol. It, then, computes inputCol1 divided by inputCol2, and adds its result as a new column to the dataset. The details of the implemeting a custom Tranfomer is explained [here](https://www.oreilly.com/learning/extend-spark-ml-for-your-own-modeltransformer-types). Please read it before before we start to implement it.First, define the given parameters of the Transformer and implement a method to validate their schemas (StructType).<jupyter_code>from pyspark.sql.types import StructField, StructType, DoubleType, StringType
from pyspark.ml.param import Param, Params

# Code<jupyter_output><empty_output><jupyter_text>Then, we extend the class Transformer, and implement its setter functions for the input and output columns, and call then setInputCol1, setInputCol2, and setOutputCol. Morever, we need to override the methods copy, transformSchema, and the transform. The details of what we need to cover in these methods is given [here](https://www.oreilly.com/learning/extend-spark-ml-for-your-own-modeltransformer-types).<jupyter_code># Code<jupyter_output><empty_output><jupyter_text>Now, an instance of MyTransformer, and set the input columns total_rooms and households, and the output column rooms_per_household and run it over the housing dataset.<jupyter_code># myTransformer = MyTransformer(inputCol1 = "total_rooms",
#                               inputCol2 = "households",
#                               outputCol = "rooms_per_household")

# myDataset = myTransformer.transform(housing).
#             select("rooms_per_household").show()<jupyter_output><empty_output><jupyter_text>8. Custom estimator (predictor) text<jupyter_code># code<jupyter_output><empty_output><jupyter_text>9.  As the last step, we are given a dataset called data/ccdefault.csv. The dataset represents default of credit card clients. It has 30,000 cases and 24 different attributes. More details about the dataset is available at data/ccdefault.txt. In this task, we should make three models, compare their results and conclude the ideal solution. Here are the suggested steps:1.   Load the data.2.   Carry out some exploratory analysis3.   Train a model to predict the target variable4.   What else can we do with this data? Anything we can do to devise a better solution?In regards to training a model, we must:*   Employ three different models (logistic regression, decision tree, and random forest).*   Compare the models' performances (e.g. AUC).*   Defend your choice of best model (e.g., what are the strengths and weaknesses of each of these models?).<jupyter_code># code<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.18558198656255676, 'fraction_numerical': 0.07493493129955814, 'mean_word_length': 2.7690201893464126, 'pattern_counts': {'":': 1, '<': 130, '<?xml version=': 0, '>': 129, 'https://': 5, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 3, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '37', 'dir': 'jupyter-structured-clean-dedup', 'id': '590376', 'license': 'MIT', 'n_tokens_mistral': 11321, 'n_tokens_neox': 9653, 'n_words': 2476, 'path': 'pySpark_lab.ipynb', 'repo_name': 'NicoEssi/Autoencoders', 'size': '33042'}