<jupyter_start><jupyter_text>Assignment 3We will use pre-populated accelerometer data to apply an Apache SparkML clustering method.<jupyter_code>!pip install pyspark==2.4.5
try:
    from pyspark import SparkContext, SparkConf
    from pyspark.sql import SparkSession
except ImportError as e:
    printmd('<<<<<!!!!! Please restart your kernel after installing Apache Spark !!!!!>>>>>')
sc = SparkContext.getOrCreate(SparkConf().setMaster("local[*]"))

spark = SparkSession \
    .builder \
    .getOrCreate()
!wget https://github.com/IBM/coursera/raw/master/coursera_ml/a2.parquet<jupyter_output>--2021-11-05 12:08:05--  https://github.com/IBM/coursera/raw/master/coursera_ml/a2.parquet
Resolving github.com (github.com)... 172.16.58.3
Connecting to github.com (github.com)|172.16.58.3|:443... connected.
HTTP request sent, awaiting response... 301 Moved Permanently
Location: https://github.com/IBM/claimed/raw/master/coursera_ml/a2.parquet [following]
--2021-11-05 12:08:05--  https://github.com/IBM/claimed/raw/master/coursera_ml/a2.parquet
Reusing existing connection to github.com:443.
HTTP request sent, awaiting response... 302 Found
Location: https://raw.githubusercontent.com/IBM/claimed/master/coursera_ml/a2.parquet [following]
--2021-11-05 12:08:05--  https://raw.githubusercontent.com/IBM/claimed/master/coursera_ml/a2.parquet
Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 172.16.58.3, 172.16.17.32, 172.16.17.32, ...
Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|172.16.58.3|:443... connected.
HTTP request sent, awaiting r[...]<jupyter_text>Now it’s time to have a look at the recorded sensor data. You should see data similar to the one exemplified below….<jupyter_code>df = spark.read.load('a2.parquet')

df.createOrReplaceTempView("df")
spark.sql("SELECT * from df").show()<jupyter_output>+-----+-----------+-------------------+-------------------+-------------------+
|CLASS|   SENSORID|                  X|                  Y|                  Z|
+-----+-----------+-------------------+-------------------+-------------------+
|    0|         26| 380.66434005495194| -139.3470983812975|-247.93697521077704|
|    0|         29| 104.74324299209692| -32.27421440203938|-25.105013725863852|
|    0| 8589934658| 118.11469236129976| 45.916682927433534| -87.97203782706572|
|    0|34359738398| 246.55394030642543|-0.6122810693132044|-398.18662513951506|
|    0|17179869241|-190.32584900181487|  234.7849657520335|-206.34483804019288|
|    0|25769803830| 178.62396382387422| -47.07529438881511|  84.38310769821979|
|    0|25769803831|  85.03128805189493|-4.3024316644854546|-1.1841857567516714|
|    0|34359738411| 26.786262674736566| -46.33193951911338| 20.880756008396055|
|    0| 8589934592|-16.203752396859194| 51.080957032176954| -96.80526656416971|
|    0|25769803852|   47.2048142440404| [...]<jupyter_text>Let’s check if we have balanced classes – this means that we have roughly the same number of examples for each class we want to predict. This is important for classification but also helpful for clustering<jupyter_code>spark.sql("SELECT count(class), class from df group by class").show()<jupyter_output>+------------+-----+
|count(class)|class|
+------------+-----+
|        1416|    1|
|        1626|    0|
+------------+-----+<jupyter_text>Let's create a VectorAssembler which consumes columns X, Y and Z and produces a column “features”<jupyter_code>from pyspark.ml.feature import VectorAssembler

vectorAssembler = VectorAssembler(inputCols=["X","Y","Z"],
                                  outputCol="features")<jupyter_output><empty_output><jupyter_text>Please instantiate a clustering algorithm from the SparkML package and assign it to the clust variable. Here we don’t need to take care of the “class” column since we are in unsupervised learning mode – so let’s pretend to not even have the “class” column for now – but it will become very handy later in assessing the clustering performance. PLEASE NOTE – IN REAL-WORLD SCENARIOS THERE IS NO CLASS COLUMN – THEREFORE YOU CAN’T ASSESS CLASSIFICATION PERFORMANCE USING THIS COLUMN<jupyter_code>from pyspark.ml.clustering import KMeans

clust = KMeans().setK(2).setSeed(1)<jupyter_output><empty_output><jupyter_text>Let’s train...<jupyter_code>from pyspark.ml import Pipeline

pipeline = Pipeline(stages=[vectorAssembler, clust])
model = pipeline.fit(df)<jupyter_output><empty_output><jupyter_text>...and evaluate...<jupyter_code>prediction = model.transform(df)
prediction.show()
prediction
pred_table1 = spark.sql('''
    select sum(correct) as correct, count(correct) as total from (
        select case when class = prediction then 1 else 0 end as correct from prediction 
    ) 
''')
pred_table1.show()
pred_table2 = spark.sql('''
    select sum(correct) as correct, count(correct) as total from (
        select case when class != prediction then 1 else 0 end as correct from prediction 
    ) 
''')
pred_table2.show()
pred_table1
pred_table1.createOrReplaceTempView('pred_table1')
pred_table2.createOrReplaceTempView('pred_table2')
pred_table1
# pred_table_union = spark.sql('''
#     select max(correct)/max(total) as accuracy from (
#         pred_table1 union pred_table2
#         )
# ''')
# pred_table_union.show()
prediction.createOrReplaceTempView('prediction')
prediction_table = spark.sql('''
select max(correct)/max(total) as accuracy from (

    select sum(correct) as correct, count(correct) as total from (
        select case when class != prediction then 1 else 0 end as correct from prediction 
    ) 
    
    union
    
    select sum(correct) as correct, count(correct) as total from (
        select case when class = prediction then 1 else 0 end as correct from prediction 
    ) 
)
''')

prediction_table.show()
# prediction_table.rdd.map(lambda row: row.accuracy).collect()[0]<jupyter_output>+------------------+
|          accuracy|
+------------------+
|0.6206443129520053|
+------------------+
{'fraction_non_alphanumeric': 0.158071560557702, 'fraction_numerical': 0.123131194355787, 'mean_word_length': 3.5071915215745646, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 30, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 6, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '529335', 'license': 'MIT', 'n_tokens_mistral': 2437, 'n_tokens_neox': 1949, 'n_words': 462, 'path': '02_Advanced_Machine_Learning_and_Signal_Processing/notebooks/week_3_pyspark_assignment_sensor_clustering.ipynb', 'repo_name': 'hsotoparada/Advanced-Data-Science-Specialization', 'size': '5969'}