<jupyter_start><jupyter_text>Spark MLLib - Linear Regression **Description** - Method to evaluate the relationship between variables.- Estimates the value of a dependent variable from the values of the independent variables.- Used when the dependent and independent variables are continuous and have some correlation.- R-Square measures how close the data is to the regression line (the R-Squar value will be between 0 and 1, and the higher the value, the better).- Input and output data are used to build the model.- The linear equation returns the values of the coefficients.- The linear equation represents the model. **Pros:** Low cost, fast and excellent for linear relations. **Cons:** It is applied only to numeric variables, and is sensitive to outliers. **Application:** It's one of the oldest models, and can be used to solve several problems.  Predicting MPG Values MPG will be the target variable and the others will be the features (predictor variables).<jupyter_code>from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.linalg import Vectors
from pyspark.ml.regression import LinearRegression
from pyspark.sql import Row
spSession = SparkSession.builder.master('local').appName('MpgPrediction').getOrCreate()
rddCars01 = sc.textFile('aux/datasets/cars.csv')<jupyter_output><empty_output><jupyter_text>**We can cache the RDD to optimize performance.**<jupyter_code>rddCars01.cache()
rddCars01.count()
rddCars01.take(5)
header = rddCars01.first()
rddCars02 = rddCars01.filter(lambda row: row != header)
rddCars02.count()<jupyter_output><empty_output><jupyter_text>Data Cleaning Analyzing the dataset, we know that there are missing values for the HP variable. So, we will use a default value for these cases.<jupyter_code>hpAvg = sc.broadcast(75.0)
def dataCleaning(strRow):
    global hpAvg
    listAttr = strRow.split(',')
    
    hpValue = hpAvg.value if listAttr[3] == '?' else listAttr[3]
    
    row = Row(
        MPG          = float(listAttr[0]),
        CYLINDERS    = float(listAttr[1]),
        DISPLACEMENT = float(listAttr[2]),
        HP           = float(hpValue),
        WEIGHT       = float(listAttr[4]),
        ACCELERATION = float(listAttr[5]),
        MODEL_YEAR   = float(listAttr[6]),
        NAME         = listAttr[7]
    )
    
    return row
rddCars03 = rddCars02.map(dataCleaning)
rddCars03.cache()
rddCars03.take(5)<jupyter_output><empty_output><jupyter_text>Exploratory Data Analysis<jupyter_code>dfCars = spSession.createDataFrame(rddCars03)
dfCars.select('MPG', 'CYLINDERS').describe().show()
for column in dfCars.columns:
    if not(isinstance(dfCars.select(column).take(1)[0][0], str)):
        print(f"MPG correlation with {column}: {dfCars.stat.corr('MPG', column)}")<jupyter_output>MPG correlation with MPG: 1.0
MPG correlation with CYLINDERS: -0.7753962854205548
MPG correlation with DISPLACEMENT: -0.8042028248058979
MPG correlation with HP: -0.774704152349872
MPG correlation with WEIGHT: -0.8317409332443347
MPG correlation with ACCELERATION: 0.42028891210165004
MPG correlation with MODEL_YEAR: 0.5792671330833091<jupyter_text>Data Pre-Processing **Dense and Sparse Vectors** Conceptually, both represent the same vector. The sparse vector usually represents the dense vector with a tuple of indices and values. When a dense vector has many zero values, it may be interesting to make it a sparse vector to use less memory. For example, take the folowing dense vector: **[1, 2, 0, 0, 5, 0, 9, 0, 0]**As a sparce vector, it would be represented as **[(0, 1, 4, 6), (1, 2, 5 , 9)]**<jupyter_code>from IPython.display import Image

Image('aux/images/vectors.png')<jupyter_output><empty_output><jupyter_text>**Creating a LabeledPoint (target, Vector[features])**It removes not relevant columns to the model (or with low correlation)<jupyter_code>def setLabeledPoint(row):
    labeledPoint = (
        row['MPG'], 
        Vectors.dense([
            row['ACCELERATION'], 
            row['DISPLACEMENT'], 
            row['HP'],
            row['WEIGHT']
        ])
    )
    
    return labeledPoint
rddCars04 = rddCars03.map(setLabeledPoint)
dfCars = spSession.createDataFrame(rddCars04, ['label', 'features'])
dfCars.select('label', 'features').show(5)
rddCars04.take(5)<jupyter_output><empty_output><jupyter_text>Machine Learning<jupyter_code>(dataTraining, dataTest) = dfCars.randomSplit([.7, .3])
dataTraining.count()
dataTest.count()
dataTraining.count() + dataTest.count() == dfCars.count()
linearRegression = LinearRegression(maxIter = 10)
model = linearRegression.fit(dataTraining)
model
print(f'Coefficients: {str(model.coefficients)}')
print(f'Intercept: {str(model.intercept)}')
predictions = model.transform(dataTest)
predictions.select('features', 'prediction').show(5)
evaluator = RegressionEvaluator(predictionCol = 'prediction', labelCol = 'label', metricName = 'r2')
evaluator.evaluate(predictions)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10961968680089486, 'fraction_numerical': 0.037014439699003455, 'mean_word_length': 4.889820359281437, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '386855', 'license': 'MIT', 'n_tokens_mistral': 1698, 'n_tokens_neox': 1557, 'n_words': 445, 'path': 'modules/09-apache-spark-machine-learning/01-spark-mllib-linear-regression.ipynb', 'repo_name': 'cfascina/rtaps', 'size': '4917'}