# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: myenv
#     language: python
#     name: myenv
# ---

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()
# -

# Mean = $\frac{1}{n} \sum_{i=1}^n a_i$

# create a rdd from 0 to 99
rdd = sc.parallelize(range(100))
sum_ = rdd.sum()
n = rdd.count()
mean = sum_/n
print(mean)

# Median <br>
# (1) sort the list <br>
# (2) pick the middle element <br>

rdd.collect()

rdd.sortBy(lambda x:x).collect()

# To access the middle element, we need to access the index.

rdd.sortBy(lambda x:x).zipWithIndex().collect()

sortedandindexed = rdd.sortBy(lambda x:x).zipWithIndex().map(lambda x:x)
n = sortedandindexed.count()
if (n%2 == 1):
    index = (n-1)/2;
    print(sortedandindexed.lookup(index))
else:
    index1 = (n/2)-1
    index2 = n/2
    value1 = sortedandindexed.lookup(index1)[0]
    value2 = sortedandindexed.lookup(index2)[0]
    print((value1 + value2)/2)

# Standard Deviation:<br>
#     - tells you how wide the is spread around the mean<br>
#     so if SD is low, all the values should be close to the mean<br>
#     - to calculate it first calculate the mean $\bar{x}$<br>
#     - SD = $\sqrt{\frac{1}{N}\sum_{i=1}^N(x_i - \bar{x})^2}$<br>

from math import sqrt
sum_ = rdd.sum()
n = rdd.count()
mean = sum_/n
sqrt(rdd.map(lambda x: pow(x-mean,2)).sum()/n)

# Skewness<br>
# - tells us how asymmetric data is spread around the mean <br>
# - check positive skew, negative skew <br>
# - Skew = $\frac{1}{n}\frac{\sum_{j=1}^n (x_j- \bar{x})^3}{\text{SD}^3}$, x_j= individual value

sd= sqrt(rdd.map(lambda x: pow(x-mean,2)).sum()/n)
n = float(n) # to round off
skw = (1/n)*rdd.map(lambda x : pow(x- mean,3)/pow(sd,3)).sum()
skw

# Kurtosis<br>
#
# - tells us the shape of the data<br>
# - indicates outlier content within the data<br>
# - kurt = $\frac{1}{n}\frac{\sum_{j=1}^n (x_j- \bar{x})^4}{\text{SD}^4}$, x_j= individual value
#

(1/n)*rdd.map(lambda x : pow(x- mean,4)/pow(sd,4)).sum()


# Covariance \& Correlation<br>
#
# - how two columns interact with each other<br>
# - how all columns interact with each other<br>
# - cov(X,Y) = $\frac{1}{n} \sum_{i=1}^n (x_i-\bar{x})(y_i -\bar{y})$

rddX = sc.parallelize(range(100))
rddY = sc.parallelize(range(100))

# to avoid loss of precision use float
meanX = rddX.sum()/float(rddX.count())
meanY = rddY.sum()/float(rddY.count())

# since we need to use rddx, rddy same time we need to zip them together
rddXY = rddX.zip(rddY)
covXY = rddXY.map(lambda x:(x[0]-meanX)*(x[1]-meanY)).sum()/rddXY.count()
covXY

# Correlation
#
# - corr(X,Y) =$ \frac{\text{cov(X,Y)}}{SD_X SD_Y}$
# <br>
#
# Measure of dependency - Correlation<br>
#  +1 Columns totally correlate<br>
#  0 columns show no interaction<br>
#  -1 inverse dependency

from math import sqrt
n = rddXY.count()
mean = sum_/n
SDX = sqrt(rdd.map(lambda x: pow(x-meanX,2)).sum()/n)
SDY = sqrt(rdd.map(lambda y: pow(y-meanY,2)).sum()/n)
corrXY = covXY/(SDX *SDY)
corrXY

# corellation matrix in practice
import random
from pyspark.mllib.stat import Statistics
col1 = sc.parallelize(range(100))
col2 = sc.parallelize(range(100,200))
col3 = sc.parallelize(list(reversed(range(100))))
col4 = sc.parallelize(random.sample(range(100),100))
data = col1
data.take(5)

data1 = col1.zip(col2)
data1.take(5)

# Welcome to exercise one of week two of “Apache Spark for Scalable Machine Learning on BigData”. In this exercise you’ll read a DataFrame in order to perform a simple statistical analysis. Then you’ll rebalance the dataset. No worries, we’ll explain everything to you, let’s get started.
#
# Let’s create a data frame from a remote file by downloading it:
#

# +
# delete files from previous runs
# !rm -f hmp.parquet*

# download the file containing the data in PARQUET format
# !wget https://github.com/IBM/coursera/raw/master/hmp.parquet
    
# create a dataframe out of it
df = spark.read.parquet('hmp.parquet')

# register a corresponding query table
df.createOrReplaceTempView('df')
# -

# This is a classical classification data set. One thing we always do during data analysis is checking if the classes are balanced. In other words, if there are more or less the same number of example in each class. Let’s find out by a simple aggregation using SQL.
#

from pyspark.sql.functions import col
counts = df.groupBy('class').count().orderBy('count')
display(counts)

df.groupBy('class').count().show()

spark.sql('select class,count(*) from df group by class').show()

# This looks nice, but it would be nice if we can aggregate further to obtain some quantitative metrics on the imbalance like, min, max, mean and standard deviation. If we divide max by min we get a measure called minmax ration which tells us something about the relationship between the smallest and largest class. Again, let’s first use SQL for those of you familiar with SQL. Don’t be scared, we’re used nested sub-selects, basically selecting from a result of a SQL query like it was a table. All within on SQL statement.
#

spark.sql('''
    select 
        *,
        max/min as minmaxratio -- compute minmaxratio based on previously computed values
        from (
            select 
                min(ct) as min, -- compute minimum value of all classes
                max(ct) as max, -- compute maximum value of all classes
                mean(ct) as mean, -- compute mean between all classes
                stddev(ct) as stddev -- compute standard deviation between all classes
                from (
                    select
                        count(*) as ct -- count the number of rows per class and rename it to ct
                        from df -- access the temporary query table called df backed by DataFrame df
                        group by class -- aggrecate over class
                )
        )   
''').show()

# The same query can be expressed using the DataFrame API. Again, don’t be scared. It’s just a sequential expression of transformation steps. You now an choose which syntax you like better.
#

df.show()
df.printSchema()

# +
from pyspark.sql.functions import col, min, max, mean, stddev

df \
    .groupBy('class') \
    .count() \
    .select([ 
        min(col("count")).alias('min'), 
        max(col("count")).alias('max'), 
        mean(col("count")).alias('mean'), 
        stddev(col("count")).alias('stddev') 
    ]) \
    .select([
        col('*'),
        (col("max") / col("min")).alias('minmaxratio')
    ]) \
    .show()

# -

# Now it’s time for you to work on the data set. First, please create a table of all classes with the respective counts, but this time, please order the table by the count number, ascending.
#

df1 = df.groupBy('class').count()
df1.sort('count',ascending=True).show()

# Pixiedust is a very sophisticated library. It takes care of sorting as well. Please modify the bar chart so that it gets sorted by the number of elements per class, ascending. Hint: It’s an option available in the UI once rendered using the display() function.
#

# + pixiedust={"displayParams": {}}
import pixiedust
from pyspark.sql.functions import col
counts = df.groupBy('class').count().orderBy('count')
display(counts)
# -

# Imbalanced classes can cause pain in machine learning. Therefore let’s rebalance. In the flowing we limit the number of elements per class to the amount of the least represented class. This is called undersampling. Other ways of rebalancing can be found here:
#
# [https://machinelearningmastery.com/tactics-to-combat-imbalanced-classes-in-your-machine-learning-dataset/](https://machinelearningmastery.com/tactics-to-combat-imbalanced-classes-in-your-machine-learning-dataset?cm_mmc=Email_Newsletter-_-Developer_Ed%2BTech-_-WW_WW-_-SkillsNetwork-Courses-IBMDeveloperSkillsNetwork-ML0201EN-SkillsNetwork-20647446&cm_mmca1=000026UJ&cm_mmca2=10006555&cm_mmca3=M12345678&cvosrc=email.Newsletter.M12345678&cvo_campaign=000026UJ)
#

# +
from pyspark.sql.functions import min

# create a lot of distinct classes from the dataset
classes = [row[0] for row in df.select('class').distinct().collect()]

# compute the number of elements of the smallest class in order to limit the number of samples per calss
min = df.groupBy('class').count().select(min('count')).first()[0]

# define the result dataframe variable
df_balanced = None

# iterate over distinct classes
for cls in classes:
    
    # only select examples for the specific class within this iteration
    # shuffle the order of the elements (by setting fraction to 1.0 sample works like shuffle)
    # return only the first n samples
    df_temp = df \
        .filter("class = '"+cls+"'") \
        .sample(False, 1.0) \
        .limit(min)
    
    # on first iteration, assing df_temp to empty df_balanced
    if df_balanced == None:    
        df_balanced = df_temp
    # afterwards, append vertically
    else:
        df_balanced=df_balanced.union(df_temp)
# -

# Please verify, by using the code cell below, if df_balanced has the same number of elements per class. You should get 6683 elements per class.
#

$$$

StarcoderdataJupyterScriptsDedupFiltered