How I built this: Machine learning with Amazon Personalize and a Customer Data Platform

Key Challenges with Machine Learning

1. Collecting and supplying quality user data to train and update your model
2. Making the insights gained from the model available where they are needed
3. Knowing how well your ML-driven experiences are working

1. Collecting and Supplying Quality User Data

1. Identity Resolution - To be able to generate recommendations based on all actions of a user you need to be able to resolve user identity. Many off-the rack ML solutions skip this requirement, tracking activity occurring on a particular device and calculating insights for that device only. This method is convenient, but it doesn't reflect a customer's true history of interaction with your brand across your digital properties and therefore can lead to incomplete insights. Identity resolution is a core capability of a CDP.
2. Consistency across platforms - Once you solve the identity resolution challenge, you still need to map data from all those different sources to a single schema that you can use to train your model. This means bringing together multiple teams of developers across multiple languages and platforms to collect data under a single schema.
3. Updating in real time - Finally you need to upload all that data to your ML platform and keep on updating it in close to real time, or your recommendations will quickly become outdated.
   

2. Making ML Insights Available and Actionable

• Without the ability to automatically connect those insights to additional systems, can your call center automation system treat high risk customers differently?
• Do your customer support representatives know when they're speaking with a high churn risk customer?
• Can your website surface retention offers?
• Can you segment on churn risk in your ESP?

3. The Project: Personalized Product Recommendations

• mParticle collects commerce data from my website, apps and stores. Each action is attributed to a master mParticle User ID and forwarded on to Amazon, using Amazon’s Kinesis streaming service.
• An AWS Lambda function converts the data into a format that can be used to train ML models and uploads it to Amazon Personalize.
• The same function requests custom product recommendations from Amazon Personalize, and uploads the recs back to a master customer profile in mParticle.
• The mParticle customer profile powers personalization on the Items4U website and apps, as well as making the same information available in my messaging and analytics platforms.

1. Collect commerce event data through mParticle
2. Create a Kinesis Stream and start streaming event data from mParticle
3. Create a Personalize dataset group
4. Create an AWS Lambda function to load data into my Personalize dataset group until I have enough data to train an ML model
5. Create a Personalize campaign
6. Update my Lambda function to request recommendations for each customer and store the recommendations on mParticle’s customer profile

Collect Data with a CDP

Capture Product Interactions

// 1. Create the product
var product = mParticle.eCommerce.createProduct(
    'Skateboard', // Name
    'prsx-10',    // SKU
    100.00,       // Price
    1             // Quantity
);

// 2. Summarize the transaction
var transactionAttributes = {
    Id: 'some-transaction-id',
    Revenue: 100,
    Tax: 9
};

// 3. Log the purchase event;
mParticle.eCommerce.logProductAction(
    mParticle.ProductActionType.Purchase,
    [product],
    null, //optional custom attributes would go here
    null, //optional custom flags would go here
    transactionAttributes);

{
    "mpid" 761556044463767215, // master user identity
	"environment": "production",
	"user_identities": {
		"email": "[email protected]"
	},
	"user_attributes": {
		"$firstname": "Milo",
		"$lastname": "Minderbinder"
	},
	"events": [{
		"data": {
			"product_action": {
				"action": "view_detail", // Others actions are "add_to_cart", "remove_from_cart", and "purchase"
				"products": [{
					"id": "prsx-10", // Product SKU
					"price": 100
				}]
			},
			"timestamp": 1604695231872
		},
		"event_type": "commerce_event"
	}]
}

Identity Resolution

Create the AWS Assets

• A Kinesis stream receives events from mParticle
• A Personalize campaign creates product recommendations
• A Lambda function acts as a broker. It transforms data from mParticle into a format accepted by Personalize, and uploads product recommendations back to mParticle.
• IAM controls access and permissions for the other components.

Create a Kinesis Stream

aws kinesis create-stream \
    --stream-name Items4UCommerceEventStream \
    --shard-count 1

StreamARN

aws iam create-role --role-name mparticle-kinesis-role --assume-role-policy-document file:///path/to/mp-trust-policy.json

aws iam put-role-policy --role-name mparticle-kinesis-role --policy-name mp-kinesis-put --policy-document file:///path/to/mp-kinesis-role.json

• Add to cart
• Add to wishlist
• Purchase
• View detail

Create a Dataset Group

aws personalize create-dataset-group --name Items4UCommerceEvents

datasetGroupArn

Create a Schema, Dataset, and Tracker

• Items: Contains detail about products, including price, category, color, etc.
• Users: Contains detail about customers, like age, location, gender, etc.
• Interactions: Details interactions between users and items. For example, a user viewing a product, purchasing it, or adding it to a cart or wishlist.

• User ID

   - this will be the mParticle ID

• Session ID

   - mParticle automatically creates a unique ID for each session, which I can use.

• Item ID

   - this will be the SKU of the product

• Event Type

   - this will be the type of product interaction: Add to Cart, Add to Wishlist, Purchase, or View Detail.

• Timestamp

   - time of the interaction. mParticle automatically records a timestamp for each interaction.

{
	"type": "record",
	"name": "Interactions",
	"namespace": "com.amazonaws.personalize.schema",
	"fields": [
		{
			"name": "USER_ID",
			"type": "string"
		},
		{
			"name": "SESSION_ID",
			"type": "string"
		},
		{
			"name": "ITEM_ID",
			"type": "string"
		},
		{
			"name": "EVENT_TYPE",
			"type": "string"
		},
		{
			"name": "TIMESTAMP",
			"type": "long"
		}
	],
	"version": "1.0"
}

aws personalize create-schema \
  --name Items4UCommerceEventSchema \
  --schema file:///path/to/items4u-commerce-event-schema.json

schemaArn

aws personalize create-dataset \
  --name Items4UCommerceEventsDataset \
  --schema-arn {{saved schema arn}} \
  --dataset-group-arn {{saved dataset group arn}} \
  --dataset-type Interactions

datasetArn

aws personalize create-event-tracker \
    --name Items4UCommerceEventTracker \
    --dataset-group-arn {{saved dataset group arn}}

trackingID

Train the Model

USER_ID,EVENT_TYPE,ITEM_ID,SESSION_ID,TIMESTAMP
761556044463767215,view_detail,prsx-23,Q8bQC4gnO8J7ewB,1595492950
-6907502341961927698,purchase,prsx-14,VA9AUJBhoJXAKr7,1595492945

1. Transform the mParticle data into my Interactions schema, and upload it to my Personalize dataset.
2. Call my Personalize Campaign and ask for updated product recommendations for the user.
3. Use the mParticle API to store the updated recommendations on the mParticle user profile.

// Import Dependencies
const AWS = require('aws-sdk');
const JSONBig = require('json-bigint')({storeAsString: true}); // needed to parse 64-bit integer MPID

// Define the product actions we want to report to Personalize
const report_actions = ["purchase", "view_detail", "add_to_cart", "add_to_wishlist"];

// Initialize Personalize
const personalizeevents = new AWS.PersonalizeEvents({apiVersion: '2018-03-22'});

exports.handler = (event, context) => {
    for (const record of event.Records) {
        // Parse encoded payload
        const payload = JSONBig.parse(Buffer.from(record.kinesis.data, 'base64').toString('ascii'));

        // Extract required params
        const events = payload.events;
        const mpid = payload.mpid;
        const sessionId = payload.message_id;
        const params = {
            sessionId: sessionId,
            userId: mpid,
            trackingId: process.env.TRACKING_ID
        };

        // Get interactions from events array
        const eventList = [];
        for (const e of events) {
            if (e.event_type === "commerce_event" && report_actions.indexOf(e.data.product_action.action) >= 0) {
                const timestamp = Math.floor(e.data.timestamp_unixtime_ms / 1000);
                const action = e.data.product_action.action;
                const event_id = e.data.event_id;
                for (const product of e.data.product_action.products) {
                    const obj = {
                        itemId: product.id,
                    };
                    eventList.push({
                        properties: obj,
                        sentAt: timestamp,
                        eventId: event_id,
                        eventType: action
                    });
                }
            }
        }
        if (eventList.length > 0) {
            params.eventList = eventList;
            // Upload interactions to tracker
            personalizeevents.putEvents(params, function(err) {
                if (err) console.log(err, err.stack);
                else console.log(`Uploaded ${eventList.length} events`)
            });
        }
    }
};

aws iam create-role \
  --role-name items4u-lambda-personalize-role \
  --assume-role-policy-document file:///path/to/lambda-trust-policy.json

Role.Arn

aws iam attach-role-policy \
  --role-name items4u-lambda-personalize-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaKinesisExecutionRole

aws iam attach-role-policy \
  --role-name items4u-lambda-personalize-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AmazonPersonalizeFullAccess

aws lambda create-function \
    --function-name Items4UPersonalizeLambda \
    --runtime nodejs12.x \
    --zip-file fileb:///path/to/Items4UPersonalizeLambda.zip \
    --role {{role arn}} \
    --handler index.handler \
    --environment Variables="{MP_KEY=SomeAccessKey,MP_SECRET=SomeAccessSecret,TRACKER_ID=SomeTrackerID}"

aws lambda create-event-source-mapping \
    --function-name Items4UPersonalizeLambda \
    --event-source-arn {{Kinesis stream arn}} \
    --starting-position LATEST

Wait...

Create an ML Campaign

• A "Solution" which describes the particular ML recipe we want to use for the campaign. One dataset group can contain many solutions.
• A "Solution Version" is an instance of a Solution trained on a specific dataset.
• The "Campaign" is what will actually dispense product recommendations for a user.

aws personalize create-solution \
  --name Items4URecsSolution \
  --dataset-group-arn {{dataset group ARN}} \
  --recipe-arn arn:aws:personalize:::recipe/aws-user-personalization

solutionArn

aws personalize create-solution-version \
  --solution-arn {{solution ARN}}

solutionVersionArn

describe-solution-version

status

ACTIVE

aws personalize describe-solution-version \
  --solution-version-arn {{solution version ARN}}

aws personalize create-campaign \
  --name Items4UProductRecsCampaign \
  --solution-version-arn arn:aws:personalize:us-east-1:521255666488:solution/Items4URecsSolution/f58f24b6 \
  --min-provisioned-tps 1

Complete the Lambda

// Import Dependencies
const AWS = require('aws-sdk');
const JSONBig = require('json-bigint')({storeAsString: true}); // needed to parse 64-bit integer MPID
const mParticle = require('mparticle');

// Define the product actions we want to report to Personalize
const report_actions = ["purchase", "view_detail", "add_to_cart", "add_to_wishlist"];

// Initialize Personalize and mParticle
const personalizeevents = new AWS.PersonalizeEvents({apiVersion: '2018-03-22'});
const personalizeruntime = new AWS.PersonalizeRuntime({apiVersion: '2018-05-22'});
const mp_api = new mParticle.EventsApi(new mParticle.Configuration(process.env.MP_KEY, process.env.MP_SECRET));

exports.handler = (event, context) => {
    for (const record of event.Records) {
        // Parse encoded payload
        const payload = JSONBig.parse(Buffer.from(record.kinesis.data, 'base64').toString('ascii'));

        // Extract required params
        const events = payload.events;
        const mpid = payload.mpid;
        const sessionId = payload.message_id;
        const params = {
            sessionId: sessionId,
            userId: mpid,
            trackingId: process.env.TRACKING_ID
        };

        // Get interactions from events array
        const eventList = [];
        for (const e of events) {
            if (e.event_type === "commerce_event" && report_actions.indexOf(e.data.product_action.action) >= 0) {
                const timestamp = Math.floor(e.data.timestamp_unixtime_ms / 1000);
                const action = e.data.product_action.action;
                const event_id = e.data.event_id;
                for (const product of e.data.product_action.products) {
                    const obj = {
                        itemId: product.id,
                    };
                    eventList.push({
                        properties: obj,
                        sentAt: timestamp,
                        eventId: event_id,
                        eventType: action
                    });
                }
            }
        }
        if (eventList.length > 0) {
            params.eventList = eventList;
            // Upload interactions to tracker
            personalizeevents.putEvents(params, function(err, data) {
                if (err) console.log(err, err.stack);
                else {
                    // Request product recs
                    var params = {
                        campaignArn: process.env.CAMPAIGN_ARN,
                        numResults: '5',
                        userId: mpid
                    };
                    personalizeruntime.getRecommendations(params, function(err, data) {
                        if (err) console.log(err, err.stack);
                        else {
                            console.log(`Uploaded ${eventList.length} events`)
                            // Upload product recs to mParticle
                            const batch = new mParticle.Batch(mParticle.Batch.Environment.development);
                            batch.mpid = mpid;
                            const itemList = [];
                            for (const item of data.itemList) {
                                itemList.push(item.itemId);
                            }
                            batch.user_attributes = {};
                            batch.user_attributes.product_recs = itemList;
                            const event = new mParticle.AppEvent(mParticle.AppEvent.CustomEventType.other, 'AWS Recs Update', {
                                product_recs: itemList.join()
                            });
                            batch.addEvent(event);
                            console.log(JSON.stringify(batch));
                            const callback = function(error, data, response) {
                                if (error) {
                                    console.error(error);
                                } else {
                                    console.log('Product Recs updated successfully');
                                }
                            };
                            mp_api.uploadEvents(batch, callback);
                        }
                    });
                }
            });
        }
    }
};

CAMPAIGN_ARN

{
    "environment": "development",
    "mpid": "-6907502341961927698",
    "user_attributes": {
        "product_recs": [
            "prsx-4",
            "prsx-2",
            "prsx-15",
            "prsx-30",
            "prsx-28"
        ]
    },
    "events": [
        {
            "data": {
                "custom_event_type": "other",
                "event_name": "AWS Recs Update",
                "custom_attributes": {
                    "product_recs": "prsx-4,prsx-2,prsx-15,prsx-30,prsx-28"
                }
            },
            "event_type": "custom_event"
        }
    ]
}

1. As an event, to record what the current recs were at a specific time.
2. As a user attribute. User attributes are kept up-to-date for each user by mParticle as new data is received. mParticle enriches incoming data with a complete set of user attributes, so any user activities captured on any platform will include the current set of recs as context.

Tracking Success

• Check to see if the current user has already been assigned to the A or B variant.
• Assign the user to a variant group, if necessary.
• Request product recs from one of two campaigns, depending on the variant.
• Record the variant info on the mParticle user profile.

const AWS = require('aws-sdk');
const JSONBig = require('json-bigint')({ storeAsString: true });
const mParticle = require('mparticle');
const trackingId = "bd973581-6505-46ae-9939-e0642a82b8b4";
const report_actions = ["purchase", "view_detail", "add_to_cart", "add_to_wishlist"];

const personalizeevents = new AWS.PersonalizeEvents({apiVersion: '2018-03-22'});
const personalizeruntime = new AWS.PersonalizeRuntime({apiVersion: '2018-05-22'});
const mp_api = new mParticle.EventsApi(new mParticle.Configuration(process.env.MP_KEY, process.env.MP_SECRET));

exports.handler = function (event, context) {
    for (const record of event.Records) {
        const payload = JSONBig.parse(Buffer.from(record.kinesis.data, 'base64').toString('ascii'));
        const events = payload.events;
        const mpid = payload.mpid;
        const sessionId = payload.message_id;
        const params = {
            sessionId: sessionId,
            userId: mpid,
            trackingId: trackingId
        };
        // Check for variant and assign one if not already assigned
        const variant_assigned = Boolean(payload.user_attributes.ml_variant); 
        const variant = variant_assigned ? payload.user_attributes.ml_variant : Math.random() > 0.5 ? "A" : "B";

        const eventList = [];
        for (const e of events) {
            if (e.event_type === "commerce_event" && report_actions.indexOf(e.data.product_action.action) >= 0) {
                const timestamp = Math.floor(e.data.timestamp_unixtime_ms / 1000);
                const action = e.data.product_action.action;
                const event_id = e.data.event_id;
                for (const product of e.data.product_action.products) {
                    const obj = {itemId: product.id,};
                    eventList.push({
                        properties: obj,
                        sentAt: timestamp,
                        eventId: event_id,
                        eventType: action
                    });
                }
            }
        }
        if (eventList.length > 0) {
            params.eventList = eventList;
            personalizeevents.putEvents(params, function(err, data) {
                if (err) console.log(err, err.stack);
                else {
                    var params = {
                      // Select campaign based on variant
                      campaignArn: process.env[`CAMPAIGN_ARN_${variant}`],
                      numResults: '5',
                      userId: mpid
                    };
                    personalizeruntime.getRecommendations(params, function(err, data) {
                      if (err) console.log(err, err.stack);
                      else {
                          const batch = new mParticle.Batch(mParticle.Batch.Environment.development);
                          batch.mpid = mpid;
                          const itemList = [];
                          for (const item of data.itemList) {
                              itemList.push(item.itemId);
                          }
                          batch.user_attributes = {};
                          batch.user_attributes.product_recs = itemList;
                          
                          // Record variant on mParticle user profile
                          if (!variant_assigned) {
                              batch.user_attributes.ml_variant = variant
                          }

                          const event = new mParticle.AppEvent(mParticle.AppEvent.CustomEventType.other, 'AWS Recs Update');
                          event.custom_attributes = {product_recs: itemList.join()};
                          batch.addEvent(event);
                          const mp_callback = function(error, data, response) {
                              if (error) {
                                  console.error(error);
                                } else {
                                  console.log('API called successfully.');
                                }
                              };
                            mp_api.uploadEvents(batch, mp_callback);
                      }
                    });
                }
            });
        }
    }
};

User journeys are not always linear. For example: a customer might look at a recommended product on my website and not buy it immediately, but pick it up later in a brick-and-mortar store, or when they next use the native app. If I'm running my experiments and analytics on a per-device basis, I'll miss that conversion.

Wrapping Up

In Machine Learning, as in all data-centric tasks, the right infrastructure is key.

• Train your model with data about your customers, rather than just a set of devices and browser sessions.
• Use centralized quality controls to train your model on the best possible dataset, that's consistent across all your data sources.
• Use scores and recommendations anywhere - not just on your website and apps, but in emails, push messages or even in your brick-and-mortar stores.
• Easily set up A/B tests and try out different models.
• Track the success of your ML experiences in your analytics platform.