<jupyter_start><jupyter_text>---title: "Inference Analysis"date: 2021-04-25type: technical_notedraft: false---  Check monitoring analysisCollect statistics, outliers and drift detections from Parquet and Kafka.<jupyter_code>from hops import hdfs
import pyarrow.parquet as pq
from hops import kafka
from hops import tls
from confluent_kafka import Producer, Consumer
import json

import pandas as pd
pd.options.display.max_columns = None
pd.options.display.max_rows = None
pd.set_option('display.max_colwidth', None)<jupyter_output>Starting Spark application<jupyter_text>Inference Statistics Read inference statistics from parquet files<jupyter_code>MONITORING_DIR = "hdfs:///Projects/" + hdfs.project_name() + "/Resources/CardFraudDetection/Monitoring/"
LOGS_STATS_DIR =  MONITORING_DIR + "credit_card_activity_stats-parquet/"
credit_card_activity_stats = spark.read.parquet(LOGS_STATS_DIR + "*.parquet")
credit_card_activity_stats.createOrReplaceTempView("credit_card_activity_stats")
desc_stats_df = spark.sql("SELECT window, feature, min, max, mean, stddev FROM credit_card_activity_stats ORDER BY window")
distr_stats_df = spark.sql("SELECT feature, distr FROM credit_card_activity_stats ORDER BY window")
corr_stats_df = spark.sql("SELECT window, feature, corr FROM credit_card_activity_stats ORDER BY window")
cov_stats_df = spark.sql("SELECT feature, cov FROM credit_card_activity_stats ORDER BY window")<jupyter_output><empty_output><jupyter_text>Descriptive statistics<jupyter_code>print(desc_stats_df.show(6, truncate=False))<jupyter_output>+------------------------------------------+-----------------+--------+-----------------+----+------+
|window                                    |feature          |min     |max              |mean|stddev|
+------------------------------------------+-----------------+--------+-----------------+----+------+
|{2021-06-08 14:21:08, 2021-06-08 14:21:14}|avg_amt_per_10m  |1.0005  |1.022727901666599|0.0 |0.04  |
|{2021-06-08 14:21:08, 2021-06-08 14:21:14}|stdev_amt_per_12h|1.002   |1.008            |0.0 |0.1   |
|{2021-06-08 14:21:08, 2021-06-08 14:21:14}|avg_amt_per_12h  |1.001   |1.0015           |0.0 |0.05  |
|{2021-06-08 14:21:08, 2021-06-08 14:21:14}|stdev_amt_per_1h |1.000745|5.264075         |0.07|1.17  |
|{2021-06-08 14:21:08, 2021-06-08 14:21:14}|avg_amt_per_1h   |1.0007  |5.791895         |0.08|0.97  |
|{2021-06-08 14:21:08, 2021-06-08 14:21:14}|num_trans_per_1h |1.001   |1.002            |0.0 |0.05  |
+------------------------------------------+-----------------+--------+-----------[...]<jupyter_text>Distributions<jupyter_code>print(distr_stats_df.show(6, truncate=False))<jupyter_output>+-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|feature          |distr                                                                                                                                                                  |
+-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|avg_amt_per_10m  |{2.0005550384521484 -> 0.0, 3.0005550384521484 -> 0.0, 4.0005550384521484 -> 0.0, 5.0005550384521484 -> 0.0, 1.0005550384521484 -> 1.0}                                |
|stdev_amt_per_12h|{3.16779012680053708 -> 0.0, 1.005739688873291 -> 13.0, 2.62727751731872556 -> 0.0, 1.54625229835510252 -> 0.0, 2.08676490783691404 -> 0.0}                            |
|avg_amt_per_12h  |{1.5285710096359253 -> 0.0, 2.56199300289[...]<jupyter_text>Correlations<jupyter_code>print(corr_stats_df.show(6, truncate=False))<jupyter_output>+------------------------------------------+-----------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|window                                    |feature          |corr                                                                                                                                                                                                              |
+------------------------------------------+-----------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|{2021-06-08 14:21:08, 2021-06-08 14:21:14}|avg_amt_per_10m  |{avg_amt_per_12h -> 0.96, stdev_amt_per_1h -> 0.0, num_trans_per_1h -> 1.06, avg_amt_per_1h -> 0.16, num_trans_per_1[...]<jupyter_text>Covariance<jupyter_code>print(cov_stats_df.show(6, truncate=False))<jupyter_output>+-----------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|feature          |cov                                                                                                                                                                                                                                                                                                                                                 |
+-----------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------[...]<jupyter_text>Outliers and Data Drift Detection (kafka)<jupyter_code>def get_consumer(topic):
    config = kafka.get_kafka_default_config()
    config['default.topic.config'] = {'auto.offset.reset': 'earliest'}
    consumer = Consumer(config)
    consumer.subscribe([topic])
    return consumer
def poll(consumer, n=2):
    df = pd.DataFrame([])
    for i in range(0, n):
        msg = consumer.poll(timeout=5.0)
        if msg is not None:
            value = msg.value()
            try: 
                d = json.loads(value.decode('utf-8'))
                df_msg = pd.DataFrame(d.items()).transpose()
                df_msg.columns = df_msg.iloc[0]
                df = df.append(df_msg.drop(df_msg.index[[0]]))
            except Exception as e:
                print("A message was read but there was an error parsing it")
                print(e)
    return df<jupyter_output><empty_output><jupyter_text>Outliers detected<jupyter_code>outliers_consumer = get_consumer("credit_card_activity_outliers")
outliers = poll(outliers_consumer, 20)
outliers.head(10)<jupyter_output>0            feature    value  type           outlier  \
1    avg_amt_per_12h  1.01958  mean  descriptiveStats   
1    avg_amt_per_12h  1.01958   min  descriptiveStats   
1     avg_amt_per_1h  1.00615  mean  descriptiveStats   
1  num_trans_per_12h   1.0015  mean  descriptiveStats   
1  num_trans_per_12h   1.0015   min  descriptiveStats   
1  num_trans_per_10m  2.05365  mean  descriptiveStats   
1  num_trans_per_10m  2.05365   min  descriptiveStats   
1    avg_amt_per_12h  1.01863  mean  descriptiveStats   
1    avg_amt_per_12h  1.01863   min  descriptiveStats   
1     avg_amt_per_1h  1.03479  mean  descriptiveStats   

0               requestTime             detectionTime  
1  2021-06-07T23:04:14.000Z  2021-06-08T12:31:42.338Z  
1  2021-06-07T23:04:14.000Z  2021-06-08T12:31:42.338Z  
1  2021-06-07T23:04:14.000Z  2021-06-08T12:31:42.339Z  
1  2021-06-07T23:04:14.000Z  2021-06-08T12:31:42.339Z  
1  2021-06-07T23:04:14.000Z  2021-06-08T12:31:42.339Z  
1  2021-06-07T23:04:14.000Z  2021-06[...]<jupyter_text>Data drift detected<jupyter_code>drift_consumer = get_consumer("credit_card_activity_drift")
drift = poll(drift_consumer, 10)
drift.head(5)<jupyter_output>0                                                                    window  \
1  {'start': '2021-06-08T10:34:32.000Z', 'end': '2021-06-08T10:34:38.000Z'}   
1  {'start': '2021-06-08T10:34:32.000Z', 'end': '2021-06-08T10:34:38.000Z'}   
1  {'start': '2021-06-08T10:34:32.000Z', 'end': '2021-06-08T10:34:38.000Z'}   
1  {'start': '2021-06-08T10:34:32.000Z', 'end': '2021-06-08T10:34:38.000Z'}   
1  {'start': '2021-06-08T10:34:32.000Z', 'end': '2021-06-08T10:34:38.000Z'}   

0            feature            drift     value             detectionTime  
1  stdev_amt_per_12h    jensenShannon  0.690542  2021-06-08T12:32:36.835Z  
1  num_trans_per_10m      wasserstein   5.26667  2021-06-08T12:32:36.836Z  
1  num_trans_per_10m  kullbackLeibler   3.39887  2021-06-08T12:32:36.837Z  
1  num_trans_per_10m    jensenShannon  0.619386  2021-06-08T12:32:36.837Z  
1  stdev_amt_per_10m      wasserstein   4.73706  2021-06-08T12:32:36.837Z
{'fraction_non_alphanumeric': 0.29257786613651426, 'fraction_numerical': 0.130329136293351, 'mean_word_length': 1.250248508946322, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 36, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '24065', 'license': 'Apache-2.0', 'n_tokens_mistral': 3345, 'n_tokens_neox': 2665, 'n_words': 416, 'path': 'mlops_world/7_inference_analysis.ipynb', 'repo_name': 'javierdlrm/events', 'size': '9054'}