<jupyter_start><jupyter_text>Music Recommender Part 3: Train Model with Debugger Hooks and Set Artifacts and Register Model----In this notebook, we'll train our model using the data we prepped with SageMaker Data Wrangler and stored in our Feature Store, attaching [SageMaker Debugger](https://docs.aws.amazon.com/sagemaker/latest/dg/train-debugger.html) to the model training so that we can capture training metrics/statistics about the model. Then, we'll log more model artifacts using [SageMaker ML Lineage Tracking](https://docs.aws.amazon.com/sagemaker/latest/dg/lineage-tracking.html). Finally we'll register the model and save its version.A machine learning training job can have problems such as system bottlenecks, overfitting, saturated activation functions, and vanishing gradients, which can compromise model performance. SageMaker Debugger profiles and debugs training jobs to help resolve such problems and improve your ML model's compute resource utilization and performance. Debugger offers tools to send alerts when training anomalies are found, take actions against the problems, and identify the root cause of them by visualizing collected metrics and tensors.Amazon SageMaker ML Lineage Tracking creates and stores information about the steps of a machine learning workflow from data preparation to model deployment. With the tracking information you can reproduce the workflow steps, track model and dataset lineage, and establish model governance and audit standards.  💡  Alert For quicker data processing times, switch to at least an ml.m5.large (General Purpose) notebook instance or larger.---- Contents- [Overview](00_overview_arch_data.ipynb)- [Part 1: Data Prep using Data Wrangler](01_music_dataprep.flow)- [Part 2a: Feature Store Creation - Tracks](02a_export_fg_tracks.ipynb)- [Part 2b: Feature Store Creation - User Preferences](02b_export_fg_5star_features.ipynb)- [Part 2c: Feature Store Creation - Ratings](02c_export_fg_ratings.ipynb)- [Part 3: Train Model with Debugger Hooks. Set Artifacts and Register Model.](03_train_model_lineage_registry_debugger.ipynb)    - [Fetch Data from Feature Store](03-feature-store)    - [Split Data and Save to S3](03-split)    - [Train Model](03-train)    - [SageMaker Debugger Reports](03-debugger)    - [Set Lineage Artifacts](03-lineage)    - [Register Model](03-register)- [Part 4: Deploy Model & Inference using Online Feature Store](04_deploy_infer_explain.ipynb)- [Part 5: Model Monitor](05_model_monitor.ipynb)- [Part 6: SageMaker Pipelines](06_pipeline.ipynb)- [Part 7: Resource Cleanup](07_clean_up.ipynb)  Load stored variablesIf you ran this notebook before, you may want to re-use the resources you aready created with AWS. Run the cell below to load any prevously created variables. You should see a print-out of the existing variables. If you don't see anything you may need to create them again or it may be your first time running this notebook.<jupyter_code>!pip install -qU smdebug
!pip install pandas --upgrade
import time
import boto3
import argparse
import pandas as pd
import pathlib
import sagemaker
import glob
import sys
import pprint
sys.path.insert(1, './code')
from parameter_store import ParameterStore
ps = ParameterStore(verbose=False)

parameters = ps.read('music-rec')
dw_ecrlist = parameters['dw_ecrlist']
fg_name_ratings = parameters['fg_name_ratings']
fg_name_tracks = parameters['fg_name_tracks']
fg_name_user_preferences = parameters['fg_name_user_preferences']
flow_export_id = parameters['flow_export_id']
flow_s3_uri = parameters['flow_s3_uri']
pretrained_model_path = parameters['pretrained_model_path']
prefix = parameters['prefix']
bucket = parameters['bucket']
ratings_data_source = parameters['ratings_data_source']
tracks_data_source = parameters['tracks_data_source']
# SHAP can take some time to install, so we will only install if it doesn't exist already
try:
    import shap
except ModuleNotFoundError:
    !conda install -y -c conda-forge shap
region = boto3.Session().region_name
boto_session = boto3.Session(region_name=region)


sagemaker_client = boto_session.client(service_name='sagemaker', region_name=region)
sagemaker_session = sagemaker.session.Session(
    boto_session=boto_session,
    sagemaker_client=sagemaker_client
)
featurestore_runtime = boto_session.client(service_name='sagemaker-featurestore-runtime', region_name=region)

feature_store_session = sagemaker.session.Session(
    boto_session=boto_session,
    sagemaker_client=sagemaker_client,
    sagemaker_featurestore_runtime_client=featurestore_runtime
)<jupyter_output><empty_output><jupyter_text>Fetch Data from Offline Feature Store [back to top](03-nb)----There are 3 feature stores for the ratings, tracks, and user preferences data. We retrieve data from all 3 before joining them.<jupyter_code>from sagemaker.feature_store.feature_group import FeatureGroup

feature_group_names = [fg_name_ratings, fg_name_tracks, fg_name_user_preferences]
feature_groups = []
for name in feature_group_names:
    feature_group = FeatureGroup(name=name, sagemaker_session=feature_store_session)
    feature_groups.append(feature_group)
s3_client = boto3.client('s3')
account_id = boto3.client('sts').get_caller_identity()["Account"]

sagemaker_role = sagemaker.get_execution_role()

s3_output_path = 's3://' + bucket
ps.add({'s3_output_path': s3_output_path}, namespace='music-rec')
feature_group_s3_prefixes = []
for feature_group in feature_groups:
    feature_group_table_name = feature_group.describe().get("OfflineStoreConfig").get("DataCatalogConfig").get("TableName")
    feature_group_s3_prefix = f'{account_id}/sagemaker/{region}/offline-store/{feature_group_table_name}'
    feature_group_s3_prefixes.append(feature_group_s3_prefix)
# wait for data to be added to offline feature store
def wait_for_offline_store(feature_group_s3_prefix):
    print(feature_group_s3_prefix)
    offline_store_contents = None
    while (offline_store_contents is None):
        objects_in_bucket = s3_client.list_objects(Bucket=bucket, Prefix=feature_group_s3_prefix)
        if ('Contents' in objects_in_bucket and len(objects_in_bucket['Contents']) > 1):
            offline_store_contents = objects_in_bucket['Contents']
        else:
            print('Waiting for data in offline store...')
            time.sleep(60)
    print('Data available.')
    
for s3_prefix in feature_group_s3_prefixes:
    wait_for_offline_store(s3_prefix)
tables = { 
    'ratings': {'feature_group': feature_groups[0],
                'cols': ['userid', 'trackid', 'rating']
               },
    'tracks': {'feature_group': feature_groups[1],
                'cols': ['trackid', 'length', 'energy', 'acousticness', 'valence', 'speechiness', 'instrumentalness', 
                        'liveness', 'tempo', 'danceability', 'genre_latin', 'genre_folk', 'genre_blues', 'genre_rap', 
                        'genre_reggae', 'genre_jazz', 'genre_rnb', 'genre_country', 'genre_electronic', 'genre_pop_rock']
               },
    'user_5star_features': {'feature_group': feature_groups[2],
                            'cols': ['userid', 'energy_5star', 'acousticness_5star', 'valence_5star', 'speechiness_5star', 'instrumentalness_5star', 
                                    'liveness_5star','tempo_5star', 'danceability_5star', 'genre_latin_5star', 'genre_folk_5star', 'genre_blues_5star', 
                                    'genre_rap_5star','genre_reggae_5star', 'genre_jazz_5star', 'genre_rnb_5star', 'genre_country_5star', 
                                    'genre_electronic_5star', 'genre_pop_rock_5star']
                           },
  }<jupyter_output><empty_output><jupyter_text>check if the athena queres have been done and the data sets exist, then just do train test split or just proceed to training<jupyter_code>def get_train_val():
    for k, v in tables.items():
        query = v['feature_group'].athena_query()
        joined_cols = ", ".join(v['cols'])
        # limit number of datapoints for training time
        query_string = "SELECT {} FROM \"{}\" LIMIT 500000".format(joined_cols, query.table_name)
        print(query_string,'\n')

        output_location = f's3://{bucket}/{prefix}/query_results/'
        query.run(query_string=query_string, output_location=output_location)
        query.wait()

        tables[k]['df'] = query.as_dataframe()  
    
    ratings = tables['ratings']['df']
    tracks = tables['tracks']['df']
    user_prefs = tables['user_5star_features']['df']
    
    print('Merging datasets...')
    print(f'Ratings: {ratings.shape}\nTracks: {tracks.shape}\nUser Prefs: {user_prefs.shape}\n')
    
    dataset = pd.merge(ratings, tracks, on='trackid', how='inner')
    dataset = pd.merge(dataset, user_prefs, on='userid', how='inner')
    dataset.drop_duplicates(inplace=True)
    dataset.drop(['userid', 'trackid'], axis=1, inplace=True)

    # split data
    from sklearn.model_selection import train_test_split
    train, val = train_test_split(dataset, test_size=0.2, random_state=42)
    print("Training dataset shape: {}\nValidation dataset shape: {}\n".format(train.shape, val.shape))

    return train, val
%%time
import pandas as pd

# create the training data if it has not been created already
if glob.glob('data/train_data.csv') and 'feature_names' in parameters:
    print('Using existing files: train_data.csv & val_data.csv')
    train = pd.read_csv('data/train_data.csv', names=["rating"]+parameters['feature_names'])
    val = pd.read_csv('data/val_data.csv', names=["rating"]+parameters['feature_names'])
else:
    print('Creating training and validation sets...\n')
    train, val = get_train_val()
    # Write to csv in S3 without headers and index column
    train.to_csv('./data/train_data.csv', header=False, index=False)
    val.to_csv('./data/val_data.csv', header=False, index=False)<jupyter_output><empty_output><jupyter_text>Save data to S3<jupyter_code>%%time
s3_client.upload_file('data/train_data.csv', bucket, f'{prefix}/data/train/train_data.csv')
s3_client.upload_file('data/val_data.csv', bucket, f'{prefix}/data/val/val_data.csv')


train_data_uri = f's3://{bucket}/{prefix}/data/train/train_data.csv'
val_data_uri = f's3://{bucket}/{prefix}/data/val/val_data.csv'
print (f"Saving training data to {train_data_uri}")

# configure data inputs for SageMaker training
from sagemaker.inputs import TrainingInput
train_input = TrainingInput(train_data_uri, content_type="text/csv")
val_input = TrainingInput(val_data_uri, content_type="text/csv")
ps.add({'train_data_uri': train_data_uri, 'val_data_uri': val_data_uri}, namespace='music-rec')<jupyter_output><empty_output><jupyter_text>Train Model  [back to top](03-nb)----<jupyter_code>from sagemaker.debugger import Rule, rule_configs
from sagemaker.debugger import DebuggerHookConfig, CollectionConfig
from sagemaker.estimator import Estimator
# variables used for parameterizing the notebook run
estimator_output_path = f's3://{bucket}/{prefix}/training_jobs'
train_instance_count = 2
train_instance_type = 'ml.m5.4xlarge'
save_interval = 2
image = sagemaker.image_uris.retrieve("xgboost", region, "0.90-2")
model_name = 'music-recommendation-model'

hyperparameters = {
    "max_depth": "4",
    "eta": "0.2",
    "objective": "reg:squarederror",
    "num_round": "100",
}
xgb_estimator = Estimator(
    role=sagemaker_role,
    instance_count=train_instance_count,
    instance_type=train_instance_type,
    image_uri=image,
    hyperparameters=hyperparameters,
    base_job_name=model_name,
    output_path=estimator_output_path,
    
    debugger_hook_config=DebuggerHookConfig(
        s3_output_path=estimator_output_path+'/debugger',  
        collection_configs=[
            CollectionConfig(
                name="metrics",
                parameters={
                    "save_interval": str(save_interval)
                }
            ),
            CollectionConfig(
                name="feature_importance",
                parameters={
                    "save_interval": str(save_interval)
                }
            ),
            CollectionConfig(
                name="full_shap",
                parameters={
                    "save_interval": str(save_interval)
                }
            ),
            CollectionConfig(
                name="average_shap",
                parameters={
                    "save_interval": str(save_interval)
                }
            ),
        ],
    ),

    rules=[
        Rule.sagemaker(
            rule_configs.loss_not_decreasing(),
            rule_parameters={
                "collection_names": "metrics",
                "num_steps": str(save_interval * 2),
            },
        )
    ],
)<jupyter_output><empty_output><jupyter_text>Train the model here in order to access SageMaker Debuggeror skip to the next notebook now to deploy a pretrained model<jupyter_code>response = sagemaker_client.list_training_jobs(
    NameContains = model_name,
    StatusEquals = 'Completed',
    SortBy='CreationTime',
    SortOrder='Descending'
)
%%time

train_model = True # True if training a new model, False if wanting to use an existing estimator once you've already trained

if train_model:
    print('Training the model')
    xgb_estimator.fit(inputs = {'train': train_input, 'validation': val_input})
    s3_debugger_output_path = xgb_estimator.latest_job_debugger_artifacts_path()
    ps.add({'s3_debugger_output_path': s3_debugger_output_path}, namespace='music-rec')
    ps.store()
elif len(response['TrainingJobSummaries']) > 0:
    training_job_name = response['TrainingJobSummaries'][0]['TrainingJobName']
    xgb_estimator = Estimator.attach(training_job_name)
    parameters = ps.read('music-rec')
    s3_debugger_output_path = parameters['s3_debugger_output_path']
    print(f'Using estimator from completed training job: {training_job_name}\nwith debugger path {s3_debugger_output_path}')
else:
    print("No existing estimator found. You'll need to run as train = True")

training_job_name = xgb_estimator.latest_training_job.job_name
print(training_job_name)

ps.add({'training_job_name': training_job_name}, namespace='music-rec')
ps.store()
import pprint
training_job_info = sagemaker_client.describe_training_job(TrainingJobName=training_job_name)
pprint.pprint(f"{training_job_info}")<jupyter_output><empty_output><jupyter_text>View SageMaker Debugger Reports [back to top](03-nb)----  Create the Trail for the Experiment uisng the Debugger to generate trial debug-outputSee this explanation for more information : https://github.com/awslabs/sagemaker-debugger/blob/master/docs/analysis.md<jupyter_code>%%time
from smdebug.trials import create_trial

try:
    trial = create_trial(s3_debugger_output_path)
except:
    parameters = ps.read('music-rec')
    s3_debugger_output_path = parameters['s3_debugger_output_path']
    trial = create_trial(s3_debugger_output_path)
feature_names = list(train.drop('rating', axis=1).columns)
print(feature_names)

ps.add({'feature_names': feature_names}, namespace='music-rec')
ps.store()
from itertools import islice
import matplotlib.pyplot as plt
import re

MAX_PLOTS = 35

def get_data(trial, tname):
    """
    For the given tensor name, walks though all the iterations
    for which you have data and fetches the values.
    Returns the set of steps and the values.
    """
    tensor = trial.tensor(tname)
    steps = tensor.steps()
    vals = [tensor.value(s) for s in steps]
    return steps, vals

def match_tensor_name_with_feature_name(tensor_name, feature_names=feature_names):
    feature_tag = tensor_name.split("/")
    for ifeat, feature_name in enumerate(feature_names):
        if feature_tag[-1]=="f{}".format(str(ifeat)): return feature_name
    return tensor_name


def plot_collection(trial, collection_name, regex='.*', figsize=(8, 6)):
    """
    Takes a `trial` and a collection name, and 
    plots all tensors that match the given regex.
    """
    fig, ax = plt.subplots(figsize=figsize)
    tensors = (trial.collection(collection_name).tensor_names)
    matched_tensors = [t for t in tensors if re.match(regex, t)]
    for tensor_name in islice(matched_tensors, MAX_PLOTS):
        steps, data = get_data(trial, tensor_name)
        ax.plot(steps, data, label=match_tensor_name_with_feature_name(tensor_name))

    ax.legend(loc='center left', bbox_to_anchor=(1, 0.5))
    ax.set_xlabel('Iteration')
plot_collection(trial, "metrics")
def plot_feature_importance(trial, importance_type="weight"):
    SUPPORTED_IMPORTANCE_TYPES = [
        "weight", "gain", "cover", "total_gain", "total_cover"]
    if importance_type not in SUPPORTED_IMPORTANCE_TYPES:
        raise ValueError(
            f"{importance_type} is not one of the supported importance types.")
    plot_collection(
        trial,
        "feature_importance",
        regex=f"feature_importance/{importance_type}/.*")
plot_feature_importance(trial, importance_type="gain")<jupyter_output><empty_output><jupyter_text>SHAP[SHAP](https://github.com/slundberg/shap) (SHapley Additive exPlanations) is another approach to explain the output of machine learning models.SHAP values represent a feature's contribution to a change in the model output. You instructed Estimator to log the average SHAP values in this example so the SHAP values (as calculated by [xgboost.predict(pred_contribs=True)](https://xgboost.readthedocs.io/en/latest/python/python_api.htmlxgboost.Booster.predict)) will be available the `average_shap` collection.![average-shap.png](attachment:average-shap.png)<jupyter_code>plot_collection(trial, "average_shap")<jupyter_output><empty_output><jupyter_text>Global explanations Global explanatory methods allow understanding the model and its feature contributions in aggregate over multiple datapoints. Here we show an aggregate bar plot that plots the mean absolute SHAP value for each feature. ![global-shap.png](attachment:global-shap.png)<jupyter_code>import shap

shap_values = trial.tensor("full_shap/f0").value(trial.last_complete_step)
shap_no_base = shap_values[:, :-1]
shap_base_value = shap_values[0, -1]
shap.summary_plot(shap_no_base, plot_type='bar', feature_names=feature_names)
shap_base_value<jupyter_output><empty_output><jupyter_text>The detailed summary plot below can provide more context over the above bar chart. It tells which features are most important and, in addition, their range of effects over the dataset. The color allows us to match how changes in the value of a feature effect the change in prediction. The 'red' indicates higher value of the feature and 'blue' indicates lower (normalized over the features).![shap-summary-plot.png](attachment:shap-summary-plot.png)<jupyter_code>X_train = train.iloc[:, 1:]
shap.summary_plot(shap_no_base, X_train)<jupyter_output><empty_output><jupyter_text>Examine Lineage [back to top](03-nb)----Though you already know the training job details from above, if we were just given the model uri, we could use SageMaker Lineage to produce the training job details which produced the model.  Data Lineage and Metrics for Best Model<jupyter_code>from sagemaker.lineage import context, artifact, association, action<jupyter_output><empty_output><jupyter_text>Training data artifact<jupyter_code>data_artifact_list = []
for data_input in training_job_info['InputDataConfig']:
    channel = data_input['ChannelName']
    data_s3_uri = data_input['DataSource']['S3DataSource']['S3Uri']

    matching_artifacts = list(artifact.Artifact.list(
        source_uri=data_s3_uri,
        sagemaker_session=sagemaker_session)
    )
    
    if matching_artifacts:
        data_artifact = matching_artifacts[0]
        print(f'Using existing artifact: {data_artifact.artifact_arn}')
    else:
        data_artifact = artifact.Artifact.create(
            artifact_name=channel,
            source_uri=data_s3_uri,
            artifact_type='Dataset',
            sagemaker_session=sagemaker_session)
        print(f'Create artifact {data_artifact.artifact_arn}: SUCCESSFUL')
    data_artifact_list.append(data_artifact)<jupyter_output><empty_output><jupyter_text>Model artifact<jupyter_code>trained_model_s3_uri = training_job_info['ModelArtifacts']['S3ModelArtifacts']

matching_artifacts = list(artifact.Artifact.list(
    source_uri=trained_model_s3_uri,
    sagemaker_session=sagemaker_session)
)

if matching_artifacts:
    model_artifact = matching_artifacts[0]
    print(f'Using existing artifact: {model_artifact.artifact_arn}')
else:
    model_artifact = artifact.Artifact.create(
        artifact_name='TrainedModel',
        source_uri=trained_model_s3_uri,
        artifact_type='Model',
        sagemaker_session=sagemaker_session)
    print(f'Create artifact {model_artifact.artifact_arn}: SUCCESSFUL')<jupyter_output><empty_output><jupyter_text>Set artifact associations<jupyter_code>trial_component = sagemaker_client.describe_trial_component(TrialComponentName=training_job_info['TrainingJobName']+'-aws-training-job')
trial_component_arn = trial_component['TrialComponentArn']<jupyter_output><empty_output><jupyter_text>Store artifacts<jupyter_code>artifact_list = data_artifact_list + [model_artifact]

for artif in artifact_list:
    if artif.artifact_type == 'Dataset':
        assoc = 'ContributedTo'
    else:
        assoc = 'Produced'
    try:
        association.Association.create(
            source_arn=artif.artifact_arn,
            destination_arn=trial_component_arn,
            association_type=assoc,
            sagemaker_session=sagemaker_session)
        print(f"Association with {artif.artifact_type}: SUCCESSFUL")
    except:
        print(f"Association already exists with {artif.artifact_type}")<jupyter_output><empty_output><jupyter_text>Model Registry [back to top](03-nb)----<jupyter_code>mpg_name = prefix+'-notebooks'

ps.add({'mpg_name':mpg_name}, namespace='music-rec')


model_packages = sagemaker_client.list_model_packages(ModelPackageGroupName=mpg_name)['ModelPackageSummaryList']

if model_packages:
    print(f'Using existing Model Package Group: {mpg_name}')
else:
    mpg_input_dict = {
        'ModelPackageGroupName': mpg_name,
        'ModelPackageGroupDescription': 'Music Recommendation Models'
    }

    mpg_response = sagemaker_client.create_model_package_group(**mpg_input_dict)
    print(f'Create Model Package Group {mpg_name}: SUCCESSFUL')
sys.path.insert(1, './code')
from inference_specification import InferenceSpecification

model_uri = training_job_info.get('ModelArtifacts', {}).get('S3ModelArtifacts')
training_image = training_job_info['AlgorithmSpecification']['TrainingImage']

mp_inference_spec = InferenceSpecification().get_inference_specification_dict(
    ecr_image=training_image,
    supports_gpu=False,
    supported_content_types=['text/csv'],
    supported_mime_types=['text/csv'])

mp_inference_spec['InferenceSpecification']['Containers'][0]['ModelDataUrl'] = model_uri
mp_input_dict = {
    'ModelPackageGroupName': mpg_name,
    'ModelPackageDescription': 'SageMaker Music Recommender',
    'ModelApprovalStatus': 'PendingManualApproval'
}

mp_input_dict.update(mp_inference_spec)
mp_response = sagemaker_client.create_model_package(**mp_input_dict)
    
model_packages = sagemaker_client.list_model_packages(ModelPackageGroupName=mpg_name)['ModelPackageSummaryList']
model_packages<jupyter_output><empty_output><jupyter_text>Create Model from Estimator<jupyter_code>model_matches = sagemaker_client.list_models(NameContains=model_name)['Models']

for model_name_match in model_matches:
    sagemaker_session.delete_model(model_name_match['ModelName'])
    print(f"Deleted existing model: {model_name_match['ModelName']}")
    
model = sagemaker_session.create_model_from_job(
    name=model_name,
    training_job_name=training_job_name,
    role=sagemaker_role,
    image_uri=training_job_info['AlgorithmSpecification']['TrainingImage']
)

print(f"Created new model: {model_name}")

ps.add({'model_name':model_name}, namespace='music-rec')
ps.store()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10183061590425754, 'fraction_numerical': 0.007631041240982445, 'mean_word_length': 4.823700825643516, 'pattern_counts': {'":': 10, '<': 69, '<?xml version=': 0, '>': 71, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '477430', 'license': 'Apache-2.0', 'n_tokens_mistral': 7630, 'n_tokens_neox': 7318, 'n_words': 1638, 'path': 'end_to_end/music_recommendation/03_train_model_lineage_registry_debugger.ipynb', 'repo_name': 'vllyakho/amazon-sagemaker-examples', 'size': '23981'}