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

# +
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
# -

# # Basic Statistics
#
# > __Note:__ marked as _experimental_
#
# ## Correlation
# - MLlib Main Guide: https://spark.apache.org/docs/2.4.3/ml-statistics.html#correlation  
# - API Docs: https://spark.apache.org/docs/2.4.3/api/python/pyspark.ml.html#pyspark.ml.stat.Correlation  
#
# Calculating the correlation between two series of data is a common operation in Statistics. Spark MLlib provides the flexibility to calculate pairwise correlations among many series.
#
# ### `ml.stat.Correlation`
#
# `Correlation` computes the correlation matrix for the input `DataFrame` of `Vector`s using the specified method. The output will be a `DataFrame` that contains the correlation matrix of the column of vectors.
#
# <img src="https://upload.wikimedia.org/wikipedia/commons/thumb/3/34/Correlation_coefficient.png/800px-Correlation_coefficient.png" alt="Correlation coefficient" width="35%"/>  
# <small>Examples of scatter diagrams with different values of correlation coefficient (ρ)<br/>Source: wikipedia</small>
#
# To use correlation, one uses the `Correlation` class from the `pyspark.ml.stat` module.  
# On this class, call the `.corr` method.
#
# Parameters for `Correlation.corr`
# - `dataset` – A DataFrame.
# - `column` – The name of the column of vectors for which the correlation coefficient needs to be computed. This must be a column of the dataset, and it must contain `Vector` objects.
# - `method` – String specifying the method to use for computing correlation. Supported: pearson (default), spearman.
#
# The supported correlation methods are currently:
# - [Pearson's correlation coefficient](https://en.wikipedia.org/wiki/Pearson_correlation_coefficient) (default)  
#     `method="pearson"`  
#     Code example:
#     ```python
#     Correlation.corr(dataframe, column="<column_of_vectors>", method="pearson")
#     ```
#
#
# - [Spearman's rank correlation coefficient](https://en.wikipedia.org/wiki/Spearman%27s_rank_correlation_coefficient)  
#     Code example
#     ```python
#     df = DataFrame.cache()
#     Correlation.corr(df, column="<column_of_vectors>", method="spearman")
#     ```
#     
# >__Note:__ For Spearman, a rank correlation, Spark needs to create an `RDD[Double]` for each column and sort it in order to retrieve the ranks and then join the columns back into an `RDD[Vector]`, which is a fairly costly operation. Hence, cache the input `DataFrame` before calling `.corr` with `method=‘spearman’` to avoid recomputing the common lineage.
#
#
#
#
# #### Example:
#

# +
from pyspark.ml.linalg import Vectors
from pyspark.ml.stat import Correlation

data = [
    (Vectors.sparse(4, [(0, 1.0), (3, -2.0)]),),
    (Vectors.dense([4.0, 5.0, 0.0, 3.0]),),
    (Vectors.dense([6.0, 7.0, 0.0, 8.0]),),
    (Vectors.sparse(4, [(0, 9.0), (3, 1.0)]),),
]
df = spark.createDataFrame(data, ["features"])

# Pearson
print("Pearson correlation matrix:")
r1 = Correlation.corr(df, "features")
r1.show()
print(f"{r1.first()[0]}\n")

# Spearman
df = df.cache()
print("Spearman correlation matrix:")
r2 = Correlation.corr(df, "features", "spearman")
r2.show()
print(f"{r2.first()[0]}\n")

# -

# ## Hypothesis Testing
#
# - MLlib Main Guide: https://spark.apache.org/docs/2.4.3/ml-statistics.html#hypothesis-testing
#     
# Hypothesis testing is a powerful tool in statistics to determine whether a result is statistically significant, whether this result occurred by chance or not. Spark's MLlib Main Guide only makes mentions of the `ChiSquareTest`, but it also supports `KolmogorovSmirnovTest`. The `KolmogorovSmirnovTest` feature was newly added in Spark 2.4.0 and has not made it to the official guide yet.
#
# ### `ml.stat.ChiSquareTest`
# - Wikipedia: [Chi-squared test](https://en.wikipedia.org/wiki/Chi-squared_test)  
# - API Docs: https://spark.apache.org/docs/2.4.3/api/python/pyspark.ml.html#pyspark.ml.stat.ChiSquareTest
#
# <p>
# <img src="https://upload.wikimedia.org/wikipedia/commons/thumb/8/8e/Chi-square_distributionCDF-English.png/600px-Chi-square_distributionCDF-English.png" alt="Chi-square distribution" width="30%" /><br/>
# <small>Chi-squared distribution, showing χ2 on the x-axis and p-value (right tail probability) on the y-axis.<br/>Source: wikipedia</small>
# </p>
#
# Conducts Pearson’s independence test for every feature against the label. For each feature, the (feature, label) pairs are converted into a contingency matrix for which the Chi-squared statistic is computed. All label and feature values must be categorical.
#
# The null hypothesis is that the occurrence of the outcomes is statistically independent.
#
# To use Chi-squared test, one uses the `ChiSquareTest` class from the `pyspark.ml.stat` module.  
# On this class, call the `.test` method.
#
# Parameters for `ChiSquareTest.test`
#
# - `dataset` – DataFrame of categorical labels and categorical features. Real-valued features will be treated as categorical for each distinct value.
# - `featuresCol` – Name of features column in dataset (must be of type `Vector`).
# - `labelCol` – Name of label column in dataset (any numerical type).
#
# Code example:
# ```python
# ChiSquareTest.test(dataframe, featuresCol="<features_column>", labelCol="<label_column>")
# ```
#
# ### `ml.stat.KolmogorovSmirnovTest`
# - Wikipedia: [Kolmogorov-Smirnov Test](https://en.wikipedia.org/wiki/Kolmogorov%E2%80%93Smirnov_test)  
# - API Docs: https://spark.apache.org/docs/2.4.3/api/python/pyspark.ml.html#pyspark.ml.stat.KolmogorovSmirnovTest
#
# <img src="https://upload.wikimedia.org/wikipedia/commons/c/cf/KS_Example.png" alt="Kolmogorov–Smirnov statistic" width="30%"/>  
# <small>Illustration of the Kolmogorov–Smirnov statistic. Red line is CDF, blue line is an ECDF, and the black arrow is the K–S statistic<br/>Source: wikipedia</small>
#
# Conduct the two-sided Kolmogorov Smirnov (KS) test for data sampled from a continuous distribution.
#
# By comparing the largest difference between the empirical cumulative distribution of the sample data and the theoretical distribution we can provide a test for the the null hypothesis that the sample data comes from that theoretical distribution.
#
# To use Kolmogorov–Smirnov, one uses the `KolmogorovSmirnovTest` class from the `pyspark.ml.stat` module.  
# On this class, call the `.test` method.
#
# Parameters for `KolmogorovSmirnovTest.test`
# (positional only)
# - `dataset` – a DataFrame containing the sample of data to test.
# - `sampleCol` – Name of sample column in dataset, of any numerical type.
# - `distName` – a string name for a theoretical distribution, (currently only support “norm”).
# - `*params` – Double values specifying the parameters to be used for the theoretical distribution. For “norm” distribution, the parameters includes mean and variance.
#     - `mean`
#     - `variance`
#
# Code example:
# ```python
# KolmogorovSmirnovTest.test(dataframe, "<sample_column>", "norm", 0.1, 1.0)
# ```
#
# ### Examples

# +
from pyspark.ml.linalg import Vectors
from pyspark.ml.stat import ChiSquareTest, KolmogorovSmirnovTest

# ChiSquareTest
data = [
    (0.0, Vectors.dense(0.5, 10.0)),
    (0.0, Vectors.dense(1.5, 20.0)),
    (1.0, Vectors.dense(1.5, 30.0)),
    (0.0, Vectors.dense(3.5, 30.0)),
    (0.0, Vectors.dense(3.5, 40.0)),
    (1.0, Vectors.dense(3.5, 40.0)),
]
df = spark.createDataFrame(data, ["label", "features"])

r = ChiSquareTest.test(df, "features", "label")
print("\nChiSquareTest:")
print("in:")
df.show()
print("out:")
r.show(1, False)
print(f" pValues: {r.first().pValues}")
print(f" degreesOfFreedom: {r.first().degreesOfFreedom}")
print(f" statistics: {r.first().statistics}")

# KolmogorovSmirnovTest
data = [[0.1], [0.15], [0.2], [0.3], [0.25]]
df = spark.createDataFrame(data, ["sample"])

r = KolmogorovSmirnovTest.test(df, 'sample', 'norm', 0.0, 1.0)
print("\nKolmogorovSmirnovTest:")
print("in:")
df.show()
print("out:")
r.show()
# Summary of the test including the p-value, test statistic
print("DataFrame based result:")
print(f" pValue: {round(r.first().pValue, 5)}")
print(f" statistic: {round(r.first().statistic, 5)}\n")
# Note that the Scala functionality of calling Statistics.kolmogorovSmirnovTest with
# a lambda to calculate the CDF is not made available in the Python API

# The RDD-based API for KolmogorovSmirnovTest outputs quite a bit nicer, which shows
# the lack of feature parity with the RDD-based API for this feature

from pyspark.mllib.stat import Statistics

# needs an rdd to run, creating an rdd with similar data
data =  [0.1, 0.15, 0.2, 0.3, 0.25]
rdd = spark.sparkContext.parallelize(data)

# run a KS test for the sample versus a standard normal distribution
testResult = Statistics.kolmogorovSmirnovTest(rdd, "norm", 0, 1)
# Note that the Scala functionality of calling Statistics.kolmogorovSmirnovTest with
# a lambda to calculate the CDF is not made available in the Python API

# summary of the test including the p-value, test statistic, and null hypothesis
# if our p-value indicates significance, we can reject the null hypothesis
print("RDD based result:")
print(testResult)
# -

# # Summarizer
#
# Spark provides vector column summary statistics for Dataframes through Summarizer. Available metrics are the column-wise max, min, mean, variance, and number of nonzeros, as well as the total count. 
#
# This feature was newly added in Spark 2.4.0 - hence it is quite 'fresh'. I found some typo's in the documentation related specifically to the function of this. Currently, the performance of this interface is about 2x-3x slower compared to using the equivalent RDD interface.
#
#
# ### `ml.stat.Summarizer`
# API guide: https://spark.apache.org/docs/2.4.3/api/python/pyspark.ml.html#pyspark.ml.stat.Summarizer  
# Tools for vectorized statistics on MLlib Vectors. The methods in this package provide various statistics for `Vectors` contained inside `DataFrame`s. This class lets users pick the statistics they would like to extract for a given column.
#
# There are two ways to use this class, singular or multiple. Instructions below.
#
# #### 1. Computing singular metrics:
# - `mean(col, weightCol=None)`  
# coefficient-wise mean.
# - `variance(col, weightCol=None)`  
# coefficient-wise variance.
# - `count(col, weightCol=None)`  
# count of all vectors seen
# - `numNonZeros(col, weightCol=None)`  
# number of non-zeros for each coefficient
# - `max(col, weightCol=None)`  
# maximum for each coefficient
# - `min(col, weightCol=None)`  
# minimum for each coefficient
# - `normL1(col, weightCol=None)`  
# L1 norm of each coefficient (sum of the absolute values)
# - `normL2(col, weightCol=None)`  
# Euclidean norm for each coefficient
#     
# #### 2. Multiple metrics  
# To compute multiple metrics, first run `Summarizer.metrics` with the specific metrics that you want to compute.
#
# - `Summarizer.metrics(*metrics)`
# Given a list of metrics, provides a builder that computes metrics from a column.  
# Available metrics (same as singular metrics above): `mean`, `variance`, `count`, `numNonZeros`, `max`, `min`, `normL1`, `normL2`
#
# This returns an instance of the `SummaryBuilder` class. On this return instance of `SummaryBuilder` use the `.summary()` method.  
# ```python
# SummaryBuilder.summary("featuresCol", weightCol=None)
# ```
#
# ### Examples

# +
from pyspark.ml.stat import Summarizer
from pyspark.ml.linalg import Vectors

data = [
    (1.0, Vectors.dense(1.0, 1.0, 1.0)),
    (0.0, Vectors.dense(1.0, 2.0, 3.0)),
]

df = spark.createDataFrame(data, ["weight", "features"])

# compute statistics for single metric "mean" with weight
df.select(Summarizer.mean(df.features, df.weight)).show(truncate=False)

# compute statistics for single metric "mean" without weight
df.select(Summarizer.mean(df.features)).show(truncate=False)

# create summarizer for multiple metrics "mean", "count", "numNonZeros"
summarizer = Summarizer.metrics("mean", "count", "numNonZeros")

# compute statistics for multiple metrics with weight
df.select(summarizer.summary(df.features, df.weight)).show(truncate=False)

# compute statistics for multiple metrics without weight
df.select(summarizer.summary(df.features)).show(truncate=False)
# -





StarcoderdataJupyterScriptsDedupFiltered