<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<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-05-04 06:41:20, 2021-05-04 06:41:26]|num_trans_per_10m|1.2   |8491.63           |197.45|1480.08|
|[2021-05-04 06:41:20, 2021-05-04 06:41:26]|avg_amt_per_10m  |1.0   |3.0               |0.05  |0.34   |
|[2021-05-04 06:41:20, 2021-05-04 06:41:26]|num_trans_per_1h |1.44  |9441.65           |219.54|1863.76|
|[2021-05-04 06:41:20, 2021-05-04 06:41:26]|avg_amt_per_12h  |3.0   |15.0              |0.28  |2.85   |
|[2021-05-04 06:41:20, 2021-05-04 06:41:26]|num_trans_per_12h|34.489|1956.6783333333333|44.7  |557.39 |
|[2021-05-04 06:41:20, 2021-05-04 06:41:26]|avg_amt_per_1h   |1.0   |3.0               |0.05  |0.67   |
+------------------------------------------+-----------------+--[...]<jupyter_text>Distributions<jupyter_code>print(distr_stats_df.show(6, truncate=False))<jupyter_output>+-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|feature          |distr                                                                                                                                                                                                                                                      |
+-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|num_trans_per_10m|[5894264.812000004 -> 0.0, 14735662.030000009 -> 0.0, 8841397.218000006 -> 0.0, 23577059.248000015 -> 0.0, 11788529.624000007 -> 0.0, 0.0 -> 43.0, 17682794.43600001 [...]<jupyter_text>Correlations<jupyter_code>print(corr_stats_df.show(6, truncate=False))<jupyter_output>+------------------------------------------+-----------------+-------------------------------------------------------------------------------------------------------------------------------------+
|window                                    |feature          |corr                                                                                                                                 |
+------------------------------------------+-----------------+-------------------------------------------------------------------------------------------------------------------------------------+
|[2021-05-04 06:41:20, 2021-05-04 06:41:26]|num_trans_per_10m|[avg_amt_per_10m -> 0.01, avg_amt_per_12h -> 0.0, num_trans_per_1h -> -0.09, avg_amt_per_1h -> -0.18, num_trans_per_12h -> 0.07]     |
|[2021-05-04 06:41:20, 2021-05-04 06:41:26]|avg_amt_per_10m  |[avg_amt_per_12h -> -0.14, num_trans_per_1h -> 0.24, avg_amt_per_1h -> -0.13, num_trans_per_12h -> -0.26, num_trans_per_10m -> 0.01] |
|[2021-05-04 06[...]<jupyter_text>Covariance<jupyter_code>print(cov_stats_df.show(6, truncate=False))<jupyter_output>+-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|feature          |cov                                                                                                                                                                                                              |
+-----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|num_trans_per_10m|[avg_amt_per_10m -> 3.0957043261568E11, avg_amt_per_12h -> -4.520635835511398E13, num_trans_per_1h -> 2.4112523343981465E15, avg_amt_per_1h -> 1.75275420088811E12, num_trans_per_12h -> 1.3834256019623732E16]  |
|avg_amt_per_10m  |[avg_amt_per_12h -> 1.226308168475E10, num_trans_per_1h -> 4.[...]<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_msg.drop('window', inplace=True, axis=1)
                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  num_trans_per_12h  1680.89   max  descriptiveStats   
1     avg_amt_per_1h        1  mean  descriptiveStats   
1    avg_amt_per_12h        9  mean  descriptiveStats   
1    avg_amt_per_12h        9   min  descriptiveStats   
1  num_trans_per_12h  264.259   max  descriptiveStats   
1     avg_amt_per_1h        2  mean  descriptiveStats   
1    avg_amt_per_12h        5  mean  descriptiveStats   
1    avg_amt_per_12h        5   min  descriptiveStats   
1  num_trans_per_12h  1331.38  mean  descriptiveStats   
1  num_trans_per_12h  1331.38   max  descriptiveStats   

0             detectionTime  
1  2021-05-04T08:53:51.427Z  
1  2021-05-04T08:53:51.429Z  
1  2021-05-04T08:53:51.429Z  
1  2021-05-04T08:53:51.429Z  
1  2021-05-04T08:53:51.429Z  
1  2021-05-04T08:53:51.430Z  
1  2021-05-04T08:53:51.430Z  
1  2021-05-04T08:53:51.430Z  
1  2021-05-04T08:53:51.430Z  
1  2021-05-04T08:53:51.430Z<jupyter_text>Data drift detected<jupyter_code>drift_consumer = get_consumer("credit_card_activity_drift")
drift = poll(drift_consumer, 20)
drift.head(5)<jupyter_output>0           feature            drift     value             detectionTime
1    avg_amt_per_1h  kullbackLeibler   7.38321  2021-05-04T10:05:23.014Z
1    avg_amt_per_1h    jensenShannon  0.690542  2021-05-04T10:05:23.014Z
1  num_trans_per_1h      wasserstein  0.766589  2021-05-04T10:05:23.015Z
1  num_trans_per_1h  kullbackLeibler  0.677257  2021-05-04T10:05:23.015Z
1  num_trans_per_1h    jensenShannon  0.182436  2021-05-04T10:05:23.015Z
{'fraction_non_alphanumeric': 0.280220424434283, 'fraction_numerical': 0.12029546253957088, 'mean_word_length': 1.3668146503884573, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 27, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '246119', 'license': 'Apache-2.0', 'n_tokens_mistral': 3189, 'n_tokens_neox': 2576, 'n_words': 410, 'path': 'data+ai_summit/6_inference_analysis.ipynb', 'repo_name': 'javierdlrm/events', 'size': '8529'}