<jupyter_start><jupyter_text>ANOVOS - Data Drift and Stability IndexFollowing notebook shows the list of "data drift and stability" related functions supported under ANOVOS package and how it can be invoked accordingly- [Data Drift Statistics](Data-Drift-Statistics)- [Attribute Stability Index](Data-Stability-Index)- [Feature Stability Estimation](Feature-Stability-Estimation)- [Appendix: Attribute Stability Index Result Visualization](Appendix:-Attribute-Stability-Index-Result-Visualization) **Setting up Spark Session**<jupyter_code>from anovos.shared.spark import *

sc.setLogLevel("ERROR")
import warnings
warnings.filterwarnings('ignore')<jupyter_output><empty_output><jupyter_text>**Input/Output Path**<jupyter_code>driftTargetPath = '../data/income_dataset/csv'
driftSourcePath = '../data/income_dataset/source'
stabilityPath = '../data/income_dataset/stability_index/'
outputPath = "../output/income_dataset/data_drift/"<jupyter_output><empty_output><jupyter_text>Data Drift Statistics- API specification of function **statistics** can be found here- Supports 5 different ways of computing the drift statistics between target and source datasets.<jupyter_code>from anovos.drift.detector import statistics
# Read source and target datasets
from anovos.data_ingest.data_ingest import read_dataset
target_df = read_dataset(spark, driftTargetPath,"csv", {'header':True,'inferSchema':True}).drop("dt_1", "dt_2")
source_df = read_dataset(spark, driftSourcePath,"csv", {'header':True,'inferSchema':True}).drop("dt_1", "dt_2")
# Example 1 - all columns + drift metric PSI
result_PSI = statistics(spark, target_df, source_df)
result_PSI.toPandas()
# Example 2 - selected columns + drift metric PSI
result_PSI = statistics(spark, target_df, source_df, 
                        list_of_cols=['age', 'education-num', 'capital-gain', 'hours-per-week'],
                        threshold=0.1)
result_PSI.toPandas()
# Example 3 - selected columns + compute statistics of JSD, HD and KS with a different bin_size
result_others = statistics(spark, target_df, source_df, 
                           list_of_cols=['age', 'education-num', 'capital-gain', 'hours-per-week'], 
                           method_type=['JSD','HD','KS'], bin_size=100)
result_others.toPandas()
# Example 4 - selected columns + save model
result_others = statistics(spark, target_df, source_df, 
                           list_of_cols=['age', 'education-num', 'capital-gain', 'hours-per-week'], 
                           pre_existing_source=False, source_path=outputPath)
result_others.toPandas()
# Example 5 - selected columns + use pre-saved model
result_others = statistics(spark, target_df, source_df, 
                           list_of_cols=['age', 'education-num', 'capital-gain', 'hours-per-week'], 
                           pre_existing_source=True, source_path=outputPath)
result_others.toPandas()<jupyter_output>2022-03-07 22:36:55.132 | DEBUG    | anovos.drift.validations:validate:20 - check the list of columns<jupyter_text>Data Stability Index - API specification of function **stability_index_computation** can be found here<jupyter_code>from anovos.drift.detector import stability_index_computation
# Read input datasets
df1 = read_dataset(spark, stabilityPath+"0","csv",{'header':True,'inferSchema':True})
df2 = read_dataset(spark, stabilityPath+"1","csv",{'header':True,'inferSchema':True})
df3 = read_dataset(spark, stabilityPath+"2","csv",{'header':True,'inferSchema':True})
df4 = read_dataset(spark, stabilityPath+"3","csv",{'header':True,'inferSchema':True})
df5 = read_dataset(spark, stabilityPath+"4","csv",{'header':True,'inferSchema':True})
df6 = read_dataset(spark, stabilityPath+"5","csv",{'header':True,'inferSchema':True})
df7 = read_dataset(spark, stabilityPath+"6","csv",{'header':True,'inferSchema':True})
df8 = read_dataset(spark, stabilityPath+"7","csv",{'header':True,'inferSchema':True})
df9 = read_dataset(spark, stabilityPath+"8","csv",{'header':True,'inferSchema':True})
df10 = read_dataset(spark, stabilityPath+"9","csv",{'header':True,'inferSchema':True})
df11 = read_dataset(spark, stabilityPath+"10","csv",{'header':True,'inferSchema':True})
df12 = read_dataset(spark, stabilityPath+"11","csv",{'header':True,'inferSchema':True})
df1.toPandas().head(5)
# Example 1 - input only one dataset
df_stability = stability_index_computation(spark, df1, appended_metric_path=outputPath+'historical_metric_1')
df_stability.toPandas() # result: should be all NaN because there is only one dataset
# Read saved intermediate statistics in Example 1
df_intermediate = read_dataset(spark, outputPath+"historical_metric_1","csv",{'header':True,'inferSchema':True})
df_intermediate.toPandas()
# Example 2 - input a few more datasets and append to existing result + nondefault threshold
df_stability = stability_index_computation(spark, df2,df3,df4,df5, 
                                           existing_metric_path=outputPath+'historical_metric_1', 
                                           appended_metric_path=outputPath+'historical_metric_1to5',threshold=2)
df_stability.toPandas()
# Example 3 - input rest of the datasets and append to previous results + nondefault threshold
df_stability = stability_index_computation(spark, df6, df7, df8, df9, df10, df11, df12, 
                                           existing_metric_path=outputPath+'historical_metric_1to5', 
                                           appended_metric_path=outputPath+'historical_metric_1to12',threshold=2)
df_stability.toPandas()
# Read saved intermediate statistics in Example 3
df_intermediate = read_dataset(spark, outputPath+"historical_metric_1to12","csv",{'header':True,'inferSchema':True})
df_intermediate.toPandas()<jupyter_output><empty_output><jupyter_text>Feature Stability Estimation- API specification of function **feature_stability_estimation** can be found here<jupyter_code>from anovos.drift.detector import feature_stability_estimation
# Example 1: estimate stability for 3 new features
odf = feature_stability_estimation(spark, df_intermediate, 
                                   {'capital_gain|hours_per_week': 'capital_gain/hours_per_week',
                                    'capital_gain|capital_loss': 'capital_gain-capital_loss',
                                    'age': 'age**2'})
odf.toPandas()
# Example 2: estimate stability for 3 new features + nondefault metric_weightages
odf = feature_stability_estimation(spark, df_intermediate, 
                                   {'capital_gain|hours_per_week': 'capital_gain/hours_per_week',
                                    'capital_gain|capital_loss': 'capital_gain-capital_loss',
                                    'age': 'age**2'},
                                   metric_weightages={"mean": 0.7, "stddev": 0.3, "kurtosis": 0})
odf.toPandas()
# Example 3: estimate stability for 3 new features + nondefault threshold
odf = feature_stability_estimation(spark, df_intermediate, 
                                   {'capital_gain|hours_per_week': 'capital_gain/hours_per_week',
                                    'capital_gain|capital_loss': 'capital_gain-capital_loss',
                                    'age': 'age**2'}, threshold=2)
odf.toPandas()<jupyter_output><empty_output><jupyter_text>Appendix: Attribute Stability Index Result VisualizationThe following function (not included in the module) can be used to visualise the trend of each metric for an attribute.<jupyter_code>def plot_attribute_result(attribute_name, df_stability_result, df_intermediate,  
                          metric_weightages={'mean':0.5,'stddev':0.3,'kurtosis':0.2}):
    """
    :params attribute_name: name of the attribute
    :params df_stability_result: output of stabilityIndex_computation
    :params df_intermediate: intermediate output of stabilityIndex_computation
    :params metric_weightages: same metric weightages used in stabilityIndex_computation
    """
    from pyspark.sql import functions as F
    metric_si_df = df_stability_result.where(F.col('attribute')==attribute_name).select('mean_si', 'stddev_si', 'kurtosis_si', 'stability_index').take(1)
    metric_si = {}
    metric_si['mean'] = metric_si_df[0]['mean_si']
    metric_si['stddev'] = metric_si_df[0]['stddev_si']
    metric_si['kurtosis'] = metric_si_df[0]['kurtosis_si']
    attribute_si = metric_si_df[0]['stability_index']
    
    metric_stats = {}
    metric_stats['mean'] = df_intermediate.where(F.col('attribute')==attribute_name).select('mean').rdd.flatMap(lambda x: x).collect()
    metric_stats['stddev'] = df_intermediate.where(F.col('attribute')==attribute_name).select('stddev').rdd.flatMap(lambda x: x).collect()
    metric_stats['kurtosis'] = df_intermediate.where(F.col('attribute')==attribute_name).select('kurtosis').rdd.flatMap(lambda x: x).collect()
    
    import matplotlib.pyplot as plt
    import math
    metrics = list(metric_weightages.keys())

    figure, axis = plt.subplots(1, len(metrics), figsize=(5*len(metrics),4))
    figure.suptitle('{}:Final Attribute Stability={}'.format(attribute_name, attribute_si))
    for i, metric in enumerate(metrics):
        metric_stability = metric_si[metric]
        metric_weightage = metric_weightages[metric]
        y = metric_stats[metric]
        axis[i].plot(list(range(1, len(y)+1)), y)
        axis[i].set_title('{}: Stability={}, Weightage={}'.format(metric, metric_stability, metric_weightage))
df_intermediate = read_dataset(spark, outputPath+"historical_metric_1to12","csv",{'header':True,'inferSchema':True})
# Example 1 - attribute age
plot_attribute_result('age', df_stability, df_intermediate)
# Example 2 - attribute education_num
plot_attribute_result('education_num', df_stability, df_intermediate)
# Example 3 - attribute logfnl
plot_attribute_result('logfnl', df_stability, df_intermediate)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12257540367624657, 'fraction_numerical': 0.013912866863003961, 'mean_word_length': 4.627428571428571, 'pattern_counts': {'":': 3, '<': 24, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '644971', 'license': 'Apache-2.0', 'n_tokens_mistral': 3059, 'n_tokens_neox': 2879, 'n_words': 576, 'path': 'examples/notebooks/drift__detector.ipynb', 'repo_name': 'ziedbouf/anovos', 'size': '9847'}