<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Computing a drift score using novelty detection modeling
This tutorial shows how to use a novelty detection model to detect skews between data splits (for example, between training and serving data). Novelty detection models can identify whether an instance belongs to a population or whether it's considered an outlier. 
The tutorial covers the following tasks
Step1:  Import libraries
Step2:  Configure Google Cloud environment settings
Step3:  Authenticate your Google Cloud account
This step is required if you run the notebook in Colab.
Step4:  Create a local workspace
Step5:  1. Download data splits
The data in this tutorial is based on the covertype dataset from UCI Machine Learning Repository. The notebook uses a version of the dataset that has been preprocessed, split, and uploaded to a public Cloud Storage bucket at the following location
Step6:  2. Train an elliptic envelope model using training data
2.1. Define metadata
Step7:  2.2. Prepare the data
Step8:  2.3. Fit the model
Step9:  3. Test the elliptic envelope model
Step10:  3.1. Generate mutated serving data
The notebook generates a dataset that contains mutated data points by shuffling each column value across the rows. This creates rows that have a random combination of feature values.
This approach makes sure that the values of each feature independently follow the distribution of the original serving data. However, the joint distribution is completely different because the feature values are generated independently.
Step11:  3.2. Use the model to score the data for drift
3.2. Compute the drift score on normal data
Step12:  3.3. Compute the drift score on mutated data
Step14:  4
Step15:  4.2. Implement the Beam pipeline
Step16:  5. Run the pipeline and display drift detection output
5.1. Configure pipeline parameter settings
Step17:  5.2. Run the pipeline
Step18:  5.3. Display the drift detection output
<ASSISTANT_TASK:>
Python Code:
!pip install -U -q apache-beam[interactive]
!pip install -U -q pandas
!pip install -U -q sklearn
# Automatically restart kernel after installs
import IPython
app = IPython.Application.instance()
app.kernel.do_shutdown(True)  
Explanation: Computing a drift score using novelty detection modeling
This tutorial shows how to use a novelty detection model to detect skews between data splits (for example, between training and serving data). Novelty detection models can identify whether an instance belongs to a population or whether it's considered an outlier. 
The tutorial covers the following tasks:
Download training and serving data splits.
Train an elliptic envelope model using the training data.
Test the model on normal and mutated datasets.
Implement an Apache Beam pipeline to compute a drift score in the request-response BigQuery data.
Run the pipeline and display drift detection output.
Setup
Install required packages
End of explanation
import os
from tensorflow import io as tf_io
import apache_beam as beam
import pandas as pd
import numpy as np
import warnings
import json
from collections import namedtuple
print("Apache Beam version: {}".format(beam.__version__))
Explanation: Import libraries
End of explanation
PROJECT_ID = "sa-data-validation"
BUCKET = "sa-data-validation"
BQ_DATASET_NAME = 'prediction_logs'
BQ_TABLE_NAME = 'covertype_classifier_logs'  
MODEL_NAME = 'covertype_classifier'
MODEL_VERSION = 'v1'
!gcloud config set project $PROJECT_ID
Explanation: Configure Google Cloud environment settings
End of explanation
try:
  from google.colab import auth
  auth.authenticate_user()
  print("Colab user is authenticated.")
except: pass
Explanation: Authenticate your Google Cloud account
This step is required if you run the notebook in Colab.
End of explanation
GCS_DATA_LOCATION = 'gs://workshop-datasets/covertype/data_validation'
WORKSPACE = './workspace'
DATA_DIR = os.path.join(WORKSPACE, 'data')
TRAIN_DATA = os.path.join(DATA_DIR, 'train.csv') 
EVAL_DATA = os.path.join(DATA_DIR, 'eval.csv') 
MODELS_DIR = os.path.join(WORKSPACE, 'models')
if tf_io.gfile.exists(WORKSPACE):
  print("Removing previous workspace artifacts...")
  tf_io.gfile.rmtree(WORKSPACE)
print("Creating a new workspace...")
tf_io.gfile.makedirs(WORKSPACE)
tf_io.gfile.makedirs(DATA_DIR)
tf_io.gfile.makedirs(MODELS_DIR)
Explanation: Create a local workspace
End of explanation
!gsutil cp gs://workshop-datasets/covertype/data_validation/training/dataset.csv {TRAIN_DATA}
!gsutil cp gs://workshop-datasets/covertype/data_validation/evaluation/dataset.csv {EVAL_DATA}
!wc -l {TRAIN_DATA}
!wc -l {EVAL_DATA}
pd.read_csv(TRAIN_DATA).head().T
Explanation: 1. Download data splits
The data in this tutorial is based on the covertype dataset from UCI Machine Learning Repository. The notebook uses a version of the dataset that has been preprocessed, split, and uploaded to a public Cloud Storage bucket at the following location: 
gs://workshop-datasets/covertype
The task in this tutorial is to predict forest cover type from cartographic variables only. 
For more information, see Cover Type Dataset.
End of explanation
TARGET_FEATURE_NAME = 'Cover_Type'
CATEGORICAL_FEATURE_NAMES = [
    'Soil_Type',
    'Wilderness_Area'
]
Explanation: 2. Train an elliptic envelope model using training data
2.1. Define metadata
End of explanation
train_data = pd.read_csv(TRAIN_DATA).drop(TARGET_FEATURE_NAME, axis=1)
from sklearn.preprocessing import OneHotEncoder
encoders = dict()
for feature_name in CATEGORICAL_FEATURE_NAMES:
  encoder = OneHotEncoder(handle_unknown='ignore')
  encoder.fit(train_data[[feature_name]])
  encoders[feature_name] = encoder
def prepare_data(data_frame):
  if type(data_frame) != pd.DataFrame:
    data_frame = pd.DataFrame(data_frame)
  
  data_frame = data_frame.reset_index()
  for feature_name, encoder in encoders.items():
    encoded_feature = pd.DataFrame(
      encoder.transform(data_frame[[feature_name]]).toarray()
    )
    data_frame = data_frame.drop(feature_name, axis=1)
    encoded_feature.columns = [feature_name+"_"+str(column) 
                               for column in encoded_feature.columns]
    data_frame = data_frame.join(encoded_feature)
  
  return data_frame
Explanation: 2.2. Prepare the data
End of explanation
prepared_training_data = prepare_data(train_data)
import time
from sklearn.covariance import EllipticEnvelope
model = EllipticEnvelope(contamination=0.)
print("Fitting...")
t0 = time.time()
model.fit(prepared_training_data)
t1 = time.time()
print("Model is fitted in {} seconds.".format(round(t1-t0)))
import statistics
training_distances = model.mahalanobis(prepared_training_data)
model._mean = statistics.mean(training_distances)
model._stdv = statistics.stdev(training_distances)
print("training distance mean: {}".format(round(model._mean, 5))) 
print("training distance stdv: {}".format(round(model._stdv, 5)))
Explanation: 2.3. Fit the model
End of explanation
def compute_drift_score(model, data_frame, stdv_units=2):
  
  distances = model.mahalanobis(data_frame)
  threshold = model._mean + (stdv_units * model._stdv)
  score = len([v for v in distances if v >= threshold]) / len(data_frame.index)
  
  return score
Explanation: 3. Test the elliptic envelope model
End of explanation
serving_data = pd.read_csv(EVAL_DATA).drop('Cover_Type', axis=1)
def shuffle_values(dataframe):     
  shuffeld_dataframe = dataframe.copy()
  for column_name in dataframe.columns:
    shuffeld_dataframe[column_name] = shuffeld_dataframe[column_name].sample(
        frac=1.0).reset_index(drop=True)
  return shuffeld_dataframe
mutated_serving_data = shuffle_values(serving_data)
mutated_serving_data.head().T
Explanation: 3.1. Generate mutated serving data
The notebook generates a dataset that contains mutated data points by shuffling each column value across the rows. This creates rows that have a random combination of feature values.
This approach makes sure that the values of each feature independently follow the distribution of the original serving data. However, the joint distribution is completely different because the feature values are generated independently.
End of explanation
stdv_units = 2
prepared_serving_data = prepare_data(serving_data)
score = compute_drift_score(model, prepared_serving_data, stdv_units)
percentage = round(score *100, 2)
print("There is {}% of the data points more than {} standard deviation units away from the mean of the training data".format(percentage, stdv_units))
Explanation: 3.2. Use the model to score the data for drift
3.2. Compute the drift score on normal data
End of explanation
prepared_mutated_data = prepare_data(mutated_serving_data)
score = compute_drift_score(model, prepared_mutated_data, stdv_units)
percentage = round(score *100, 2)
print("There is {}% of the data points more than {} standard deviation units away from the mean of the training data".format(percentage, stdv_units))
Explanation: 3.3. Compute the drift score on mutated data
End of explanation
from collections import defaultdict
def parse_batch_data(log_records):
  data_dict = defaultdict(list)
  for log_record in log_records:
    raw_data = json.loads(log_record['raw_data'])
    for raw_instance in raw_data['instances']:
      for name, value in raw_instance.items():
        data_dict[name].append(value[0])
  return data_dict
def score_data(data, model, stdv_units=2):
  distances = model.mahalanobis(data)
  threshold = model._mean + (stdv_units * model._stdv)
  outlier_count = len([v for v in distances if v >= threshold])
  records_count = len(data)
  return {'outlier_count': outlier_count, 'records_count': records_count}
def aggregate_scores(items):
  outlier_count = 0 
  records_count = 0
  for item in items:
    outlier_count += item['outlier_count']
    records_count += item['records_count']
  return {'outlier_count': outlier_count, 'records_count': records_count}
def get_query(bq_table_fullname, model_name, model_version, start_time, end_time):
  query = 
  SELECT raw_data
  FROM {}
  WHERE model = '{}'
  AND model_version = '{}'
  .format(bq_table_fullname, model_name, model_version, start_time, end_time)
  return query
Explanation: 4: Implement an Apache Beam pipeline to compute a drift score in request-response BigQuery data
This pipeline performs the following steps:
1. Reads and parses the data from a request-response logs table in BigQuery.
2. Uses the elliptic envelope novelty detection model to identify outliers.
3. Computes the percentage of the data points that are detected as outliers to generate the drift score.
4.1. Prepare helper functions
End of explanation
def run_pipeline(args):
  options = beam.options.pipeline_options.PipelineOptions(**args)
  args = namedtuple("options", args.keys())(*args.values())
  query = get_query(
      args.bq_table_fullname, args.model_name, 
      args.model_version, 
      args.start_time, 
      args.end_time
  )
  print("Starting the Beam pipeline...")
  with beam.Pipeline(options=options) as pipeline:
    (
        pipeline 
        | 'ReadBigQueryData' >> beam.io.Read(
            beam.io.BigQuerySource(query=query, use_standard_sql=True))
        | 'BatchRecords' >> beam.BatchElements(
            min_batch_size=100, max_batch_size=1000)
        | 'InstancesToBeamExamples' >> beam.Map(parse_batch_data)
        | 'PrepareData' >> beam.Map(prepare_data)
        | 'ScoreData' >> beam.Map(
            lambda data: score_data(data, args.drift_model, stdv_units=1))
        | 'CombineResults' >> beam.CombineGlobally(aggregate_scores)
        | 'ComputeRatio' >> beam.Map(
            lambda result: {
                "outlier_count": result['outlier_count'], 
                "records_count": result['records_count'],
                "drift_ratio": result['outlier_count'] / result['records_count']
                })
         | 'WriteOutput' >> beam.io.WriteToText(
             file_path_prefix=args.output_file_path, num_shards=1, shard_name_template='')
    )
    
Explanation: 4.2. Implement the Beam pipeline
End of explanation
from datetime import datetime
job_name = 'drift-detection-{}'.format(
    datetime.utcnow().strftime('%y%m%d-%H%M%S'))
bq_table_fullname = "{}.{}.{}".format(
    PROJECT_ID, BQ_DATASET_NAME, BQ_TABLE_NAME)
runner = 'InteractiveRunner'
output_dir = os.path.join(WORKSPACE, 'output')
output_path = os.path.join(output_dir, 'drift_output.json')
start_time = '2020-07-05 00:00:00 UTC'
end_time = '2020-07-06 23:59:59 UTC'
args = {
    'job_name': job_name,
    'runner': runner,
    'bq_table_fullname': bq_table_fullname,
    'model_name': MODEL_NAME,
    'model_version': MODEL_VERSION,
    'start_time': start_time,
    'end_time': end_time,
    'output_file_path': output_path,
    'project': PROJECT_ID,
    'reference_schema': reference_schema,
    'drift_model': model
}
Explanation: 5. Run the pipeline and display drift detection output
5.1. Configure pipeline parameter settings
End of explanation
!rm -r {output_dir}
print("Running pipeline...")
%time run_pipeline(args)
print("Pipeline is done.")
!ls {output_dir}
Explanation: 5.2. Run the pipeline
End of explanation
dirft_results = json.loads(open(output_path).read()).items()
for key, value in dirft_results:
  if key == 'drift_ratio':
    value = str(round(value * 100, 2)) +'%'
  print(key,':', value)
Explanation: 5.3. Display the drift detection output
End of explanation
<END_TASK>