<jupyter_start><jupyter_text>K-Means & PCAThere are **7 Questions** in this assignment. All are mandatory. Do the assignment on Google CoLab. Write your code in the cells for each question and submit this notebook on LMS.   Setup Let's setup Spark on your Colab environment.  Run the cell below!<jupyter_code># !pip install pyspark
# !pip install -U -q PyDrive
# !apt install openjdk-8-jdk-headless -qq
# import os
# os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"<jupyter_output><empty_output><jupyter_text>Now we import some of the libraries usually needed by our workload.<jupyter_code>import findspark
findspark.init()
import pyspark
from pyspark.sql import *
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark import SparkContext, SparkConf

from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline<jupyter_output><empty_output><jupyter_text>Let's initialize the Spark context.<jupyter_code># create the session
conf = SparkConf()

# create the context
sc = pyspark.SparkContext(conf=conf)
spark = SparkSession.builder.getOrCreate()<jupyter_output><empty_output><jupyter_text>You can easily check the current version and get the link of the web interface. In the Spark UI, you can monitor the progress of your job and debug the performance bottlenecks (if your Colab is running with a **local runtime**).<jupyter_code>spark<jupyter_output><empty_output><jupyter_text>Data Preprocessing In this Colab, rather than downloading a file from Google Drive, we will load a famous machine learning dataset, the [Breast Cancer Wisconsin dataset](https://scikit-learn.org/stable/modules/generated/sklearn.datasets.load_breast_cancer.html), using the ```scikit-learn``` datasets loader.<jupyter_code>from sklearn.datasets import load_breast_cancer
breast_cancer = load_breast_cancer()<jupyter_output><empty_output><jupyter_text>For convenience, given that the dataset is small, we first construct a Pandas dataframe, tune the schema, and then convert it into a Spark dataframe.<jupyter_code>pd_df = pd.DataFrame(breast_cancer.data, columns=breast_cancer.feature_names)
df = spark.createDataFrame(pd_df)

def set_df_columns_nullable(spark, df, column_list, nullable=False):
    for struct_field in df.schema:
        if struct_field.name in column_list:
            struct_field.nullable = nullable
    df_mod = spark.createDataFrame(df.rdd, df.schema)
    return df_mod

df = set_df_columns_nullable(spark, df, df.columns)
df = df.withColumn('features', array(df.columns))
vectors = df.rdd.map(lambda row: Vectors.dense(row.features))

df.printSchema()<jupyter_output>root
 |-- mean radius: double (nullable = false)
 |-- mean texture: double (nullable = false)
 |-- mean perimeter: double (nullable = false)
 |-- mean area: double (nullable = false)
 |-- mean smoothness: double (nullable = false)
 |-- mean compactness: double (nullable = false)
 |-- mean concavity: double (nullable = false)
 |-- mean concave points: double (nullable = false)
 |-- mean symmetry: double (nullable = false)
 |-- mean fractal dimension: double (nullable = false)
 |-- radius error: double (nullable = false)
 |-- texture error: double (nullable = false)
 |-- perimeter error: double (nullable = false)
 |-- area error: double (nullable = false)
 |-- smoothness error: double (nullable = false)
 |-- compactness error: double (nullable = false)
 |-- concavity error: double (nullable = false)
 |-- concave points error: double (nullable = false)
 |-- symmetry error: double (nullable = false)
 |-- fractal dimension error: double (nullable = false)
 |-- worst radius: double (nullable[...]<jupyter_text>With the next cell, we build the two datastructures that we will be using throughout this Colab:*   ```features```, a dataframe of Dense vectors, containing all the original features in the dataset;*   ```labels```, a series of binary labels indicating if the corresponding set of features belongs to a subject with breast cancer, or not.<jupyter_code>from pyspark.ml.linalg import Vectors
features = spark.createDataFrame(vectors.map(Row), ["features"])
labels = pd.Series(breast_cancer.target)<jupyter_output><empty_output><jupyter_text>**Question01:** If you run successfully the Setup and Data Preprocessing stages, you are now ready to cluster the data with the [K-means](https://spark.apache.org/docs/latest/ml-clustering.html) algorithm included in MLlib (Spark's Machine Learning library).Set the ```k``` parameter to **2**, fit the model, and the compute the [Silhouette score](https://en.wikipedia.org/wiki/Silhouette_(clustering)) (i.e., a measure of quality of the obtained clustering).  **IMPORTANT:** use the MLlib implementation of the Silhouette score (via ```ClusteringEvaluator```).<jupyter_code>def train_KMeans(features, k):
    # trains a k-means model.
    kmeans = KMeans().setK(k).setSeed(1)
    model = kmeans.fit(features)
    
    # make predictions
    predictions = model.transform(features)

    # evaluate clustering by computing Silhouette score
    evaluator = ClusteringEvaluator()

    silhouette = evaluator.evaluate(predictions)
    
    return model, predictions, silhouette

# call to train
model, predictions, silhouette = train_KMeans(features, 2)

print("Silhouette with squared euclidean distance = " + str(silhouette))<jupyter_output>Silhouette with squared euclidean distance = 0.8342904262826144<jupyter_text>**Question02:** Compute the within cluster sum of squares (WSS) using your **OWN** method in Spark. Note that the score can be be obtained using the `summary.trainingCost` method of the fitted kmeans model.<jupyter_code># get cluster centers to compute distances from
centers = model.clusterCenters() 

# evaluate clustering by computing Within Set Sum of Squared Errors
def error(row):
    point = row['features'].toArray()
    cluster = row['prediction']
    
    center = centers[cluster]
    
    distances = point - center
    
    return np.sum(np.square(distances))

WSS = predictions.rdd.map(lambda row: error(row)).reduce(lambda x, y: x + y)

print("Within Cluster Sum of Squared Error (Own) = " + str(WSS))
print("Within Cluster Sum of Squared Error (Summary) = " + str(model.summary.trainingCost))<jupyter_output>Within Cluster Sum of Squared Error (Own) = 77943099.87829885
Within Cluster Sum of Squared Error (Summary) = 77943099.87829879<jupyter_text>**Question03:** **Take** the predictions produced by K-means, and compare them with the ```labels``` variable (i.e., the ground truth from our dataset).  Compute how many data points in the dataset have been clustered correctly (i.e., positive cases in one cluster, negative cases in the other).*HINT*: you can use ```np.count_nonzero(series_a == series_b)``` to quickly compute the element-wise comparison of two series.**IMPORTANT**: K-means is a clustering algorithm, so it will not output a label for each data point, but just a cluster identifier!  As such, label ```0``` does not necessarily match the cluster identifier ```0```.<jupyter_code>cluster_id_label_mapping = {1:0, 0:1}

def get_accuracy(predictions, labels):
    
    predictions_pdf = predictions.select('prediction').toPandas()
    
    predictions_pdf['predict_label_adj'] = predictions_pdf['prediction'].map(cluster_id_label_mapping)
    
    return np.count_nonzero(labels == predictions_pdf.predict_label_adj)

get_accuracy(predictions, labels)<jupyter_output><empty_output><jupyter_text>**Question03:** Now perform dimensionality reduction on the ```features``` using the [PCA](https://spark.apache.org/docs/latest/ml-features.htmlpca) statistical procedure, available as well in MLlib.Set the ```k``` parameter to **2**, effectively reducing the dataset size of a **15X** factor.<jupyter_code>from pyspark.ml.feature import PCA


pca = PCA(k=2, inputCol="features", outputCol="pca")
model = pca.fit(features)

transformed = model.transform(features)

pca_features = transformed.select('pca').withColumnRenamed('pca', 'features')<jupyter_output><empty_output><jupyter_text>**Question04:** Now run K-means with the same parameters as above, but on the ```pcaFeatures``` produced by the PCA reduction you just executed.<jupyter_code># train model with pca features
model, predictions, silhouette = train_KMeans(pca_features, 2)<jupyter_output><empty_output><jupyter_text>**Question05:** Compute the Silhouette score, as well as the number of data points that have been clustered correctly.<jupyter_code>acc = get_accuracy(predictions, labels)
print("Silhouette with squared euclidean distance = " + str(silhouette))
print("Data points that have been clustered correctly = " + str(acc))<jupyter_output>Silhouette with squared euclidean distance = 0.8348610363444835
Data points that have been clustered correctly = 486<jupyter_text>**Question06:** Visualize the dataset by plotting a scatter plot of the two PCA components. You need to plot two scatter plots:1) Highlight the two **actual** labels in the dataset2) Highlight the two **clusters** found by K-Means in the dataset<jupyter_code>pdf = predictions.toPandas()
pdf['label'] = labels
pdf['predict_label_adj'] = pdf['prediction'].map(cluster_id_label_mapping)
# separate out coords
pdf['x'] = pdf['features'].apply(lambda f:f[0])
pdf['y'] = pdf['features'].apply(lambda f:f[1])
# predictions
plt.scatter(pdf['x'], pdf['y'], c=pdf['prediction'], s=50, cmap='viridis');
# labels
plt.scatter(pdf['x'], pdf['y'], c=labels, s=50, cmap='viridis');<jupyter_output><empty_output><jupyter_text>Use of `cluster_id_label_mapping` dictionary to assess accuracy makes even more sense by looking at these graphs above, however by applying that adjustment we have below comparison<jupyter_code># predictions, adjusted labels
plt.scatter(pdf['x'], pdf['y'], c=pdf['predict_label_adj'], s=50, cmap='viridis');<jupyter_output><empty_output><jupyter_text>**Question07:** Repeat the process of Question01 for K = 1 to 10. **Plot** separately1) the Sihouette score for each K2) the within cluster sum of squares (WSS) for each K<jupyter_code>ks = [x for x in range(2, 11)]
sils = []
wsss = []

for k in ks:
    
    model, predictions, silhouette = train_KMeans(features, k)
    
    sils.append(silhouette)
    wsss.append(model.summary.trainingCost)    

# silhouette
plt.plot(ks, sils)
# wss
plt.plot(ks, wsss)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11051419800460476, 'fraction_numerical': 0.011895625479662318, 'mean_word_length': 5.564861460957179, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 61, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '282598', 'license': 'Apache-2.0', 'n_tokens_mistral': 3304, 'n_tokens_neox': 3137, 'n_words': 1018, 'path': 'k-mean-pca-pyspark/kmean-pca.ipynb', 'repo_name': 'azharmunir43/pyspark-practice', 'size': '10424'}