<jupyter_start><jupyter_text>1. Importing dependencies<jupyter_code>import os

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns

from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession, SQLContext

from pyspark.sql.types import *
import pyspark.sql.functions as F
from pyspark.sql.functions import udf, col

from pyspark.ml.regression import LinearRegression
from pyspark.mllib.evaluation import RegressionMetrics

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator, CrossValidatorModel
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.evaluation import RegressionEvaluator
# Visualization
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = "all"

pd.set_option('display.max_columns', 200)
pd.set_option('display.max_colwidth', 400)

from matplotlib import rcParams
sns.set(context='notebook', style='whitegrid', rc={'figure.figsize': (18,4)})
rcParams['figure.figsize'] = 18,4

%matplotlib inline
%config InlineBackend.figure_format = 'retina'
# setting random seed for notebook reproducability
rnd_seed=23
np.random.seed=rnd_seed
np.random.set_state=rnd_seed<jupyter_output><empty_output><jupyter_text>2. Creating the Spark Session<jupyter_code>spark = SparkSession.builder.master("local").appName("Abalone-Spark-Test").getOrCreate()
spark
sc = spark.sparkContext
sc
sqlContext = SQLContext(spark.sparkContext)
sqlContext<jupyter_output><empty_output><jupyter_text>3. Load The Data From a File Into a Dataframe<jupyter_code>FILENAME = '../../data/raw/abalone.csv'<jupyter_output><empty_output><jupyter_text>Specifying the schema when loading data into a DataFrame will give better performance than schema inference.<jupyter_code># define the schema, corresponding to a line in the csv data file.
schema = StructType([
    StructField("sex", StringType(), nullable=True),
    StructField("length", FloatType(), nullable=True),
    StructField("diameter", FloatType(), nullable=True),
    StructField("height", FloatType(), nullable=True),
    StructField("whole_weight", FloatType(), nullable=True),
    StructField("shucked_weight", FloatType(), nullable=True),
    StructField("viscera_weight", FloatType(), nullable=True),
    StructField("shell_weight", FloatType(), nullable=True),
    StructField("rings", IntegerType(), nullable=True)]
)
# Load housing data
abalone_df = spark.read.csv(path=FILENAME, schema=schema, header=True).cache()
# Show first five rows
abalone_df.show(5)
# show the dataframe columns
abalone_df.columns
# show the schema of the dataframe
abalone_df.printSchema()<jupyter_output>root
 |-- sex: string (nullable = true)
 |-- length: float (nullable = true)
 |-- diameter: float (nullable = true)
 |-- height: float (nullable = true)
 |-- whole_weight: float (nullable = true)
 |-- shucked_weight: float (nullable = true)
 |-- viscera_weight: float (nullable = true)
 |-- shell_weight: float (nullable = true)
 |-- rings: integer (nullable = true)<jupyter_text>4. Data Exploration<jupyter_code>(abalone_df.describe().select(
                    "summary",
                    F.round("length", 4).alias("length"),
                    F.round("diameter", 4).alias("diameter"),
                    F.round("height", 4).alias("height"),
                    F.round("whole_weight", 4).alias("whole_weight"),
                    F.round("shucked_weight", 4).alias("shucked_weight"),
                    F.round("viscera_weight", 4).alias("viscera_weight"),
                    F.round("shell_weight", 4).alias("shell_weight"),
                    F.round("rings", 4).alias("rings"),
                    ).show())
df_pandas =  abalone_df.toPandas()
for col in abalone_df.columns:
    plt.title(col)
    df_pandas[col].hist(figsize=(6, 4))
    plt.show()<jupyter_output><empty_output><jupyter_text>5. Data Preprocessing  6. Feature Engineering<jupyter_code>featureCols = ['length', 'diameter', 'height','whole_weight',
               'shucked_weight','viscera_weight','shell_weight']
# put features into a feature vector column
assembler = VectorAssembler(inputCols=featureCols, outputCol="features")
assembled_df = assembler.transform(abalone_df)
assembled_df.show(10, truncate=False)
# Initialize the `standardScaler`
standardScaler = StandardScaler(inputCol="features", outputCol="features_scaled")
# Fit the DataFrame to the scaler
prep_df = standardScaler.fit(assembled_df).transform(assembled_df)
# Inspect the result
prep_df.select("features", "features_scaled").show(10, truncate=False)<jupyter_output>+----------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+
|features                                                                                                                                      |features_scaled                                                                                                                          |
+----------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+
|[0.45500001311302185,0.36500000953674316,0.0949999988079071,0.5139999985694885,0.22450000047683716,0.10100000351667404,0.15000000596046448]   |[3.7887[...]<jupyter_text>7. Building A Machine Learning Model With Spark ML<jupyter_code># Split the data into train and test sets
train_data, test_data = prep_df.randomSplit([.8,.2], seed=rnd_seed)
# Initialize `lr`
lr = (LinearRegression(featuresCol='features_scaled', labelCol="rings", predictionCol='predrings',
                       maxIter=10, regParam=0.3, elasticNetParam=0.8, standardization=False))
# Fit the data to the model
linearModel = lr.fit(train_data)<jupyter_output><empty_output><jupyter_text>8. Evaluating the Model<jupyter_code># Coefficients for the model
linearModel.coefficients
featureCols
# Intercept for the model
linearModel.intercept
coeff_df = pd.DataFrame({"Feature": ["Intercept"] + featureCols, "Co-efficients": np.insert(linearModel.coefficients.toArray(), 0, linearModel.intercept)})
coeff_df = coeff_df[["Feature", "Co-efficients"]]
coeff_df
# Generate predictions
predictions = linearModel.transform(test_data)
# Extract the predictions and the "known" correct labels
predandlabels = predictions.select("predrings", "rings")
predandlabels.show()
# Get the RMSE
print("RMSE: {0}".format(linearModel.summary.rootMeanSquaredError))
print("MAE: {0}".format(linearModel.summary.meanAbsoluteError))
# Get the R2
print("R2: {0}".format(linearModel.summary.r2))
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1918520658769142, 'fraction_numerical': 0.026148511990754117, 'mean_word_length': 2.9741676234213545, 'pattern_counts': {'":': 2, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 11, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '544176', 'license': 'MIT', 'n_tokens_mistral': 2089, 'n_tokens_neox': 1939, 'n_words': 408, 'path': 'notebooks/lucas/ntbk01_abalone_pyspark.ipynb', 'repo_name': 'dwelcaslu/IFSP-Projeto-Interdisciplinar-2021.02', 'size': '6922'}