<jupyter_start><jupyter_text>Environment Setup<jupyter_code>!pip install pyspark==3.0.1 py4j==0.10.9<jupyter_output>Collecting pyspark==3.0.1
[?25l  Downloading https://files.pythonhosted.org/packages/f0/26/198fc8c0b98580f617cb03cb298c6056587b8f0447e20fa40c5b634ced77/pyspark-3.0.1.tar.gz (204.2MB)
[K     |████████████████████████████████| 204.2MB 91kB/s 
[?25hCollecting py4j==0.10.9
[?25l  Downloading https://files.pythonhosted.org/packages/9e/b6/6a4fb90cd235dc8e265a6a2067f2a2c99f0d91787f06aca4bcf7c23f3f80/py4j-0.10.9-py2.py3-none-any.whl (198kB)
[K     |████████████████████████████████| 204kB 23.1MB/s 
[?25hBuilding wheels for collected packages: pyspark
  Building wheel for pyspark (setup.py) ... [?25l[?25hdone
  Created wheel for pyspark: filename=pyspark-3.0.1-py2.py3-none-any.whl size=204612242 sha256=1e09689bb531ee39777d99c5c88b007d217243171e1f026ff5266f066d373c26
  Stored in directory: /root/.cache/pip/wheels/5e/bd/07/031766ca628adec8435bb40f0bd83bb676ce65ff4007f8e73f
Successfully built pyspark
Installing collected packages: py4j, pyspark
Successfully installed py4j-0.10.9 pyspark-3.0[...]<jupyter_text>Download Data<jupyter_code>!wget https://archive.ics.uci.edu/ml/machine-learning-databases/abalone/abalone.data<jupyter_output>--2021-07-01 06:54:05--  https://archive.ics.uci.edu/ml/machine-learning-databases/abalone/abalone.data
Resolving archive.ics.uci.edu (archive.ics.uci.edu)... 192.168.127.12
Connecting to archive.ics.uci.edu (archive.ics.uci.edu)|192.168.127.12|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 191873 (187K) [application/x-httpd-php]
Saving to: ‘abalone.data’

abalone.data        100%[===================>] 187.38K   308KB/s    in 0.6s    

2021-07-01 06:54:06 (308 KB/s) - ‘abalone.data’ saved [191873/191873]<jupyter_text>TaskThe task for the project is to take **Abalone Dataset**, perofrm Singular Value Decompsition on Spark Clsuter and Visualize the result in reduced dimension.  Spark Session<jupyter_code># Starting Spark Session

from pyspark.sql import SparkSession
from pyspark import SparkConf
config = {
    "spark.kubernetes.namespace": "spark",
    "spark.kubernetes.container.image": "itayb/spark:3.1.1-hadoop-3.2.0-aws",
    "spark.executor.instances": "2",
    "spark.executor.memory": "512m",
    "spark.executor.cores": "1",
    "spark.driver.blockManager.port": "36782",
    "spark.driver.port": "35472",
    "spark.driver.host": "jupyter.spark.svc.cluster.local",
    "spark.driver.bindAddress": "0.0.0.0",
    "spark.hadoop.fs.s3a.endpoint": "localstack.kube-system.svc.cluster.local:35716",
    "spark.hadoop.fs.s3a.connection.ssl.enabled": "false",
    "spark.hadoop.fs.s3a.path.style.access": "true",
    "spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
    "spark.hadoop.fs.s3a.access.key": "ayush",
    "spark.hadoop.fs.s3a.secret.key": "ayush",
    "spark.hadoop.com.amazonaws.services.s3.enableV4": "true",
}
def get_spark_session(app_name: str, master: string):
    conf = SparkConf()
    conf.setMaster(master)
    for key, value in config.items():
        conf.set(key, value)    
    return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
try:
    spark = get_spark_session("SVD-on-Spark", "k8s://https://kubernetes.default.svc.cluster.local")
except:
    spark = SparkSession.builder\
            .master("spark://spark-master:7077")\
            .appName('SVD-on-Spark')\
            .config('spark.executor.memory', '512m')\
            .getOrCreate()
from pyspark.sql.types import *
from pyspark.sql.functions import col, lit, when, min, max, avg
from pyspark.sql import Window
from pyspark.sql import functions as f

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt<jupyter_output><empty_output><jupyter_text>Data Loading<jupyter_code># DEfining table Schema Manually

data_schema = [
               StructField('Sex', StringType(), True),
               StructField('Length', DoubleType(), True),
               StructField('Diameter', DoubleType(), True),
               StructField('Height', DoubleType(), True),
               StructField('Whole_weight', DoubleType(), True),
               StructField('Shucked_weight', DoubleType(), True),
               StructField('Viscera_weight', DoubleType(), True),
               StructField('Shell_weight', DoubleType(), True),
               StructField('Rings', IntegerType(), True)
            ]
# Loading data in Spark environemnt.

final_struc = StructType(fields = data_schema)

try:
    data = spark.read.csv(
                          's3://abalone.data',
                          sep = ',',
                          header = True,
                          schema = final_struc 
)
    
except:
    data = spark.read.csv(
                          'abalone.data',
                          sep = ',',
                          header = True,
                          schema = final_struc 
)

data.printSchema()
# Data Subset

data.show(5)
# Count of Entries in Dataset

data.count()
# Summary Statistics

data.describe().show()<jupyter_output>+-------+----+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-----------------+
|summary| Sex|            Length|           Diameter|             Height|      Whole_weight|     Shucked_weight|     Viscera_weight|       Shell_weight|            Rings|
+-------+----+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-----------------+
|  count|4176|              4176|               4176|               4176|              4176|               4176|               4176|               4176|             4176|
|   mean|null|0.5240086206896557|0.40789152298850717|0.13952705938697216|0.8288175287356346| 0.3593997844827576|0.18061266762452136|0.23885213122605353|9.932471264367816|
| stddev|null|0.1201025466927038|0.09924953111992146|0.04182639038077335|0.4904235478411498|0.22197971440458067|0.10962045433929166| 0.1392125497[...]<jupyter_text>Singualar Value Decomposition<jupyter_code>from pyspark.mllib.linalg import Vectors
from pyspark.mllib.linalg.distributed import RowMatrix
# Seprating Categoerical and Numerical Data

Y = data.select("Sex")
X = data.select(['Length', 'Diameter', 'Height', 'Whole_weight', 'Shucked_weight', 'Viscera_weight', 'Shell_weight', 'Rings'])
# Converting Dataframe into RDD

data_matrix = RowMatrix(X.rdd.map(list))
# Performing Singlular Value decomposition

svd = data_matrix.computeSVD(len(X.columns), computeU=True)

U = svd.U
sigma = svd.s
V = svd.V<jupyter_output><empty_output><jupyter_text>Visualization<jupyter_code># Cummualtive Sum of Contribution of Each Singular Vectors

csum = np.cumsum(sigma)/sigma.sum()

plt.figure(figsize=(12, 5))
plt.plot(range(1, len(sigma)+1), csum)
plt.scatter(range(1, len(sigma)+1), csum)
plt.xlabel("Count of Singular Vectors")
plt.ylabel("Information Content in %")
for i, val in enumerate(csum):
    plt.annotate(np.round(100*val,3), [i+1, val-0.005])
plt.ylim(0.92, 1.01)
plt.xlim(0.5, 8.75)
plt.title("Singualar Vectors Information Content")
plt.show()
# Collect first two Singualr Vectors

Xs1 = np.array(U.rows.map(lambda x: x[0]).collect())
Xs2 = np.array(U.rows.map(lambda x: x[1]).collect())
color_dict = {"M": "red", "F": "green", "I": "dodgerblue"}
name_dict =  {"M": "Male", "F": "Female", "I": "Infant"}
# Fetching Sex attrubute for each data point.

Y_dense = np.array(list(map(lambda x: x["Sex"], Y.collect())))
# Colour Allocation to each Sex.

color = np.array(list(map(lambda x: color_dict[x["Sex"]], Y.collect())))
# Original Data Visualization

pd.plotting.scatter_matrix(X.toPandas(), figsize=(15, 15), c=color, diagonal='kde')
plt.show()
# Data Vsualization in Transformed Space

plt.figure(figsize=(16, 6))
jitter = 0

for key in color_dict.keys():
    idx = (Y_dense == key)
    plt.scatter(Xs1[idx]+jitter, Xs2[idx], color=color[idx], label=name_dict[key], alpha=0.6)
    jitter += 0.0001
plt.xlabel("New Basis 1")
plt.ylabel("New Basis 2")
plt.legend()
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.18841821743388834, 'fraction_numerical': 0.08900587659157688, 'mean_word_length': 3.0662020905923346, 'pattern_counts': {'":': 22, '<': 22, '<?xml version=': 0, '>': 23, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '88826', 'license': 'MIT', 'n_tokens_mistral': 3402, 'n_tokens_neox': 2893, 'n_words': 499, 'path': 'Notebooks/SVD_on_Spark.ipynb', 'repo_name': 'itsayushthada/SVD-on-Spark', 'size': '8168'}