Monitoring Logs with the Elastic Stack - Create Kibana Dashboard for Model Inputs

Author: ChristopherGS

packages/ml_api/api/controller.py

@@ -54,9 +54,8 @@ def predict():
     if request.method == "POST":
         # Step 1: Extract POST data from request body as JSON
         json_data = request.get_json()
-        _logger.info(
-            f'Inputs for model: {ModelType.LASSO.name} '
-            f'Input values: {json_data}')
+        for entry in json_data:
+            _logger.info(entry)
 
         # Step 2a: Get and save live model predictions
         persistence = PredictionPersistence(db_session=current_app.db_session)

packages/ml_api/docker/kibana/config/kibana_example_inputs_dashboard.ndjson

@@ -8,10 +8,19 @@ input {
 }
 
 output {
-	elasticsearch {
-		hosts => "elasticsearch:9200"
-		user => "elastic"
-		password => "changeme"
- 		index => "webapp_logs-%{+YYYY.MM.dd}"
+	if [LotArea] {
+		elasticsearch {
+			hosts => "elasticsearch:9200"
+			user => "elastic"
+			password => "changeme"
+			index => "input_logs-%{+YYYY.MM.dd}"
+		}
+	} else {
+		elasticsearch {
+			hosts => "elasticsearch:9200"
+			user => "elastic"
+			password => "changeme"
+ 			index => "webapp_logs-%{+YYYY.MM.dd}"
  		}
-}
+	}
+}

packages/ml_api/docker/logstash/pipeline/logstash.conf

@@ -2,7 +2,7 @@
 import os
 import time
 import typing as t
-from random import randint
+from random import randint, choice
 
 import pandas as pd
 import requests
@@ -19,14 +19,21 @@
 
 SECOND_FLR_SF_MAP = {"min": 0, "max": 1862}
 
+BSMT_QUAL_VALUES = ('Gd', 'TA', 'Ex', 'Fa')
+
 
 def _generate_random_int(value: int, value_ranges: t.Mapping) -> int:
     """Generate random integer within a min and max range."""
     random_value = randint(value_ranges["min"], value_ranges["max"])
-
     return int(random_value)
 
 
+def _select_random_category(value: str, value_options: t.Sequence) -> str:
+    """Select random category given a sequence of categories."""
+    random_category = choice(value_options)
+    return random_category
+
+
 def _prepare_inputs(dataframe: pd.DataFrame) -> pd.DataFrame:
     """Prepare input data by removing key rows with NA values."""
     clean_inputs_df = dataframe.dropna(
@@ -43,6 +50,10 @@ def _prepare_inputs(dataframe: pd.DataFrame) -> pd.DataFrame:
         _generate_random_int, value_ranges=LOT_AREA_MAP
     )
 
+    clean_inputs_df.loc[:, "BsmtQual"] = clean_inputs_df["BsmtQual"].apply(
+        _select_random_category, value_options=BSMT_QUAL_VALUES
+    )
+
     return clean_inputs_df
 
 
@@ -74,6 +85,7 @@ def populate_database(n_predictions: int = 500, anomaly: bool = False) -> None:
         clean_inputs_df.loc[:, "OverallQual"] = 1
         clean_inputs_df.loc[:, "GrLivArea"] = 1
 
+    clean_inputs_df = clean_inputs_df.where(pd.notnull(clean_inputs_df), None)
     for index, data in clean_inputs_df.iterrows():
         if index > n_predictions:
             if anomaly:

packages/ml_api/scripts/populate_database.py

@@ -98,7 +98,7 @@ passenv =
 
 setenv =
   PYTHONPATH=.
-  DB_HOST={env:DB_HOST:localhost}
+  DB_HOST=localhost
 
 commands = python scripts/populate_database.py {posargs}