<jupyter_start><jupyter_text>Amazon Personalize User Segmentation Workshop - Lab 7In this workshop we will prepare and execute a segmentation job that creates multiple audiences/segments of users with an affinity for different product categories in the Retail Demo Store. We could then take these user segments and use them to target users with promotions for items in each category.  SetupThe workshop will be using the python programming language and the AWS SDK for python. Even if you are not fluent in python, the code cells should be reasonably intuitive. In practice, you can use any programming language supported by the AWS SDK to complete the same steps from this workshop in your application environment. Update dependenciesTo get started, we need to perform a bit of setup. First, we need to ensure that a current version of botocore is locally installed. The botocore library is used by boto3, the AWS SDK library for python. We need a current version to be able to access some of the newer Amazon Personalize features.The following cell will update pip and install the latest botocore library.<jupyter_code>import sys
!{sys.executable} -m pip install --upgrade pip
!{sys.executable} -m pip install --upgrade --no-deps --force-reinstall botocore<jupyter_output><empty_output><jupyter_text>Import dependencies and prepare clientsFirst we will import the libraries and create the clients needed for this workshop.<jupyter_code># Import dependencies
import boto3
import json
import pandas as pd
import numpy
import time
import requests
import botocore
from datetime import datetime
from IPython.display import display, HTML
from packaging import version

# Create clients
personalize = boto3.client('personalize')
servicediscovery = boto3.client('servicediscovery')
ssm = boto3.client('ssm')<jupyter_output><empty_output><jupyter_text>The following cell will load the saved variables from the earlier foundational Personalize workshops.<jupyter_code>%store -r<jupyter_output><empty_output><jupyter_text>Lookup working S3 bucket nameWe will stage the segmentation job input file on S3 and have the segmentation job output written to S3. We'll use the same S3 stack bucket used for other Personalize workshops. The bucket name is stored in SSM so let's lookup the value.<jupyter_code>bucketresponse = ssm.get_parameter(
    Name='retaildemostore-stack-bucket'
)

# We will use this bucket to store our training data:
bucket = bucketresponse['Parameter']['Value']     # Do Not Change

print('Bucket: {}'.format(bucket))<jupyter_output><empty_output><jupyter_text>Retrieve IP addresses of Products and Users microservicesWe will select a random product from the Retail Demo Store's catalog to use to find users with an affinity for the product's attributes. Let's get the local IP address of the Products microservice so we can call its API to retrieve products.<jupyter_code>response = servicediscovery.discover_instances(
    NamespaceName='retaildemostore.local',
    ServiceName='products',
    MaxResults=1,
    HealthStatus='HEALTHY'
)

assert len(response['Instances']) > 0, 'Products service instance not found; check ECS to ensure it launched cleanly'

products_service_instance = response['Instances'][0]['Attributes']['AWS_INSTANCE_IPV4']
print('Products Service Instance IP: {}'.format(products_service_instance))<jupyter_output><empty_output><jupyter_text>We will also lookup info on each user included in the user segmentation output job to see who was recommended. We'll use the Users microservice to get details on each user ID.<jupyter_code>response = servicediscovery.discover_instances(
    NamespaceName='retaildemostore.local',
    ServiceName='users',
    MaxResults=1,
    HealthStatus='HEALTHY'
)

assert len(response['Instances']) > 0, 'Users service instance not found; check ECS to ensure it launched cleanly'

users_service_instance = response['Instances'][0]['Attributes']['AWS_INSTANCE_IPV4']
print('Users Service Instance IP: {}'.format(users_service_instance))<jupyter_output><empty_output><jupyter_text>Load products into DataFrameLet's load all of the products from the Products microservice into a Pandas dataframe.<jupyter_code>response = requests.get('http://{}/products/all'.format(products_service_instance))
products = response.json()
products_df = pd.DataFrame(products)
pd.set_option('display.max_rows', 5)

products_df<jupyter_output><empty_output><jupyter_text>Prepare input file for batch segment jobNext we will prepare a user segmentation input file by randomly selecting a product from the catalog.First, let's consider the format of the job input file. Below is a sample of the input file for an item affinity job that builds 3 user segments looking for users for a Video On Demand application that are interested in both comedies and action movies, users just interested in comedies, and users interested in both horror and action movies:```javascript{"itemAttributes": "ITEMS.genres = \"Comedy\" AND ITEMS.genres = \"Action\""}{"itemAttributes": "ITEMS.genres = \"Comedy\""}{"itemAttributes": "ITEMS.genres = \"Horror\" AND ITEMS.genres = \"Action\""}```For our user segmentation job, we will select a few categories and use each category name as the item attribute we want to use to group users.<jupyter_code>categories = products_df['category'].unique()
print(categories)<jupyter_output><empty_output><jupyter_text>Let's randomly select 3 categories to use in the job input file.<jupyter_code>segment_categories = numpy.random.choice(categories, 3, False)
print(segment_categories)
# Create and write job input file to disk
json_input_filename = "item_affinity_json_input.json"
with open(json_input_filename, 'w') as json_input:
    for category in segment_categories:
        # Write line that specifies the query for users with an affinity for the CATEGORY_L1 field
        json_input.write(f'{{"itemAttributes": "ITEMS.CATEGORY_L1 = \\"{category}\\""}}\n')<jupyter_output><empty_output><jupyter_text>Display the job input file contents. One very important characteristic of the job input file is that the `itemAttributes` query expression for each segment must be fully defined in a single line.<jupyter_code>!cat $json_input_filename<jupyter_output><empty_output><jupyter_text>Upload job input file to S3 bucketBefore we can create a segmentation job, we have to upload the job input file to our S3 bucket.<jupyter_code># Upload file to S3
boto3.Session().resource('s3').Bucket(bucket).Object(json_input_filename).upload_file(json_input_filename)
s3_input_path = "s3://" + bucket + "/" + json_input_filename
print(s3_input_path)<jupyter_output><empty_output><jupyter_text>Define job output locationWe also need to define an output location in our S3 bucket where the segmentation job writes its output.<jupyter_code># Define the output path
s3_output_path = "s3://" + bucket + "/user-segmentation/attrib-affinity/"
print(s3_output_path)<jupyter_output><empty_output><jupyter_text>Create batch segment jobFinally, we're ready to create a batch segment job. There are several required parameters including the solution version ARN for the item affinity model we created in a prior workshop, the IAM role that Personalize needs to access the job input file and write the output file, and the job input and output locations. We're also optionally specifying that we only want the top 25 users in our user segment.<jupyter_code>response = personalize.create_batch_segment_job (
    solutionVersionArn = item_attribute_affinity_solution_version_arn,
    jobName = "retaildemostore-item-attrib-affinity_" + str(round(time.time()*1000)),
    roleArn = role_arn,
    jobInput = {"s3DataSource": {"path": s3_input_path}},
    jobOutput = {"s3DataDestination":{"path": s3_output_path}},
    numResults = 25
)
job_arn = response['batchSegmentJobArn']
print(json.dumps(response, indent=2, default=str))<jupyter_output><empty_output><jupyter_text>Wait for batch segment job to completeThe user segmentation job can take several minutes to complete. Even though our input file only specifies a few item affinity query expressions, there is a certain amount of fixed overhead required for Personalize to spin up the compute resources needed to execute the job. This overhead is amortized for larger input files that generate many user segments.<jupyter_code>%%time 

current_time = datetime.now()
print("Import Started on: ", current_time.strftime("%I:%M:%S %p"))

max_time = time.time() + 3*60*60 # 3 hours
while time.time() < max_time:
    response = personalize.describe_batch_segment_job(
        batchSegmentJobArn = job_arn
    )
    status = response["batchSegmentJob"]['status']
    print("DatasetSegmentJob: {}".format(status))
    
    if status == "ACTIVE" or status == "CREATE FAILED":
        break
        
    time.sleep(60)
    
current_time = datetime.now()
print("Import Completed on: ", current_time.strftime("%I:%M:%S %p"))<jupyter_output><empty_output><jupyter_text>Download and inspect job output fileLet's download the output file from S3 and display its contents.<jupyter_code>s3 = boto3.client('s3')
out_file = json_input_filename + ".out"
export_name = 'user-segmentation/attrib-affinity/' + out_file
s3.download_file(bucket, export_name, out_file)

!cat $out_file<jupyter_output><empty_output><jupyter_text>Notice that the input item attribute query expressions are echoed in the output file but we also have `output` and `error` elements for each user segment. The `output` element has a `usersList` array that contains the user IDs for the segment. If there were any errors enountered while generating a segment, details will be included in the `error` element for the segment.  Inspect user segmentsLet's take a look at the users selected for each segment to get a sense of their interests, recent behavior, and affinity for the category used for each segment.To be able to determine the relevance of the users for each segment, we will display the last 5 products viewed for the users in each segment. We'll get the 5 most recent products viewed from the interactions dataset used to train the item affinity solution version.Let's start by loading the interactions dataset into a dataframe and display the first few rows.<jupyter_code>interactions_df = pd.read_csv('interactions.csv')
interactions_df['USER_ID'] = interactions_df.USER_ID.astype(str)
interactions_df['TIMESTAMP'] = interactions_df.TIMESTAMP.astype(int)
interactions_df.head(5)<jupyter_output><empty_output><jupyter_text>We query this dataframe for the user ID included in each segment to fetch each user's interactions.Next we'll create a utility function that we'll call for each user in a segment that will lookup the user name from the Users microservice as well as the most recent product view events. This information is concatenated to a working dataframe that collects info on all users in a user segment.<jupyter_code>def append_user_info_and_recent_interactions(df_work, user_id: str, count: int = 5):
    # Fetch user info from the Users microservice.
    response = requests.get('http://{}/users/id/{}'.format(users_service_instance, user_id))
    user = response.json()

    d = {
        'userId': [ user_id ], 
        'name': [ user['first_name'] + ' ' + user['last_name'] ]
    }

    # Lookup product view events for the user, sort descending by timestamp, and trim list.
    user_interactions_df = interactions_df.loc[
        (interactions_df['USER_ID'] == user_id) & (interactions_df['EVENT_TYPE'] == 'View')
    ]
    user_interactions_df = user_interactions_df.sort_values(by=['TIMESTAMP'], ascending=False)
    user_interactions_df = user_interactions_df.head(count)
    
    # Lookup product name and category for recently viewed products.
    prod_count = 1
    for product_id in user_interactions_df['ITEM_ID']:
        product = products_df.loc[products_df['id'] == product_id].iloc[0]
        d[f'productView{prod_count}'] = [ f'{product["name"]} ({product["category"]})' ]
        prod_count += 1
    
    # Concatenate user details and recent views to working dataframe.
    return pd.concat([df_work, pd.DataFrame(data=d)], axis=0, ignore_index=True)<jupyter_output><empty_output><jupyter_text>Finally, we're ready to read the contents of the user segmentation job output file and write an intuitive overview of each segment.<jupyter_code>pd.options.display.max_rows = 25

with open(out_file) as segments_file:
    # Read all lines from the segmentation output file.
    segment_lines = segments_file.readlines()
    
    for idx, segment_line in enumerate(segment_lines):
        segment = json.loads(segment_line)
        attrib = segment['input']['itemAttributes']
        users_list = segment['output']['usersList']

        # Get info on each user in the segment.
        segment_info_df = pd.DataFrame()
        for user_id in users_list:
            segment_info_df = append_user_info_and_recent_interactions(segment_info_df, str(user_id))

        # Display details on all users in the segment
        display(HTML(f'<h3>Segment {idx + 1}: {attrib}</h3>'))
        display(segment_info_df)
        display(HTML('<hr/>'))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08238787786492043, 'fraction_numerical': 0.005710804842762506, 'mean_word_length': 5.378824672170957, 'pattern_counts': {'":': 9, '<': 77, '<?xml version=': 0, '>': 78, 'https://': 0, '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': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '316210', 'license': 'MIT-0', 'n_tokens_mistral': 3720, 'n_tokens_neox': 3595, 'n_words': 1367, 'path': 'workshop/1-Personalization/Lab-7-Intelligent-user-segmentation-using-Amazon-Personalize.ipynb', 'repo_name': 'xtrim-serverless/retail-demo-store', 'size': '13133'}