{
"cells": [
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"# Build a Customer Churn Model for Music Streaming App Users: Overview and Data Preparation\n"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"This notebook's CI test result for us-west-2 is as follows. CI test results in other regions can be found at the end of the notebook. \n",
"\n",
"\n",
"\n",
"---"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"\n",
"## Background\n",
"\n",
"This notebook is one of a sequence of notebooks that show you how to use various SageMaker functionalities to build, train, and deploy the model from end to end, including data pre-processing steps like ingestion, cleaning and processing, feature engineering, training and hyperparameter tuning, model explainability, and eventually deploy the model. There are two parts of the demo: \n",
"\n",
"1. Build a Customer Churn Model for Music Streaming App Users: Overview and Data Preparation (current notebook) - you will process the data with the help of Data Wrangler, then create features from the cleaned data. By the end of part 1, you will have a complete feature data set that contains all attributes built for each user, and it is ready for modeling.\n",
"1. Build a Customer Churn Model for Music Streaming App Users: Model Selection and Model Explainability - you will use the data set built from part 1 to find an optimal model for the use case, then test the model predictability with the test data. \n",
"\n",
"For how to set up the SageMaker Studio Notebook environment, please check the [onboarding video]( https://www.youtube.com/watch?v=wiDHCWVrjCU&feature=youtu.be). And for a list of services covered in the use case demo, please check the documentation linked in each section.\n",
"\n",
"\n",
"## Content\n",
"* [Overview](#Overview)\n",
"* [Data Selection](#Data-Selection)\n",
"* [Ingest Data](#Ingest-Data)\n",
"* [Data Cleaning and Data Exploration](#Data-Cleaning)\n",
"* [Pre-processing with SageMaker Data Wrangler](#Pre-processing-with-SageMaker-Data-Wrangler)\n",
"* [Feature Engineering with SageMaker Processing](#Feature-Engineering-with-SageMaker-Processing)\n",
"* [Data Splitting](#Data-Splitting)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"## Overview\n",
"\n",
"### What is Customer Churn and why is it important for businesses?\n",
"Customer churn, or customer retention/attrition, means a customer has the tendency to leave and stop paying for a business. It is one of the primary metrics companies want to track to get a sense of their customer satisfaction, especially for a subscription-based business model. The company can track churn rate (defined as the percentage of customers churned during a period) as a health indicator for the business, but we would love to identify the at-risk customers before they churn and offer appropriate treatment to keep them with the business, and this is where machine learning comes into play.\n",
"\n",
"### Use Cases for Customer Churn\n",
"\n",
"Any subscription-based business would track customer churn as one of the most critical Key Performance Indicators (KPIs). Such companies and industries include Telecom companies (cable, cell phone, internet, etc.), digital subscriptions of media (news, forums, blogposts platforms, etc.), music and video streaming services, and other Software as a Service (SaaS) providers (e-commerce, CRM, Mar-Tech, cloud computing, video conference provider, and visualization and data science tools, etc.)\n",
"\n",
"### Define Business problem\n",
"\n",
"To start with, here are some common business problems to consider depending on your specific use cases and your focus:\n",
"\n",
" * Will this customer churn (cancel the plan, cancel the subscription)?\n",
" * Will this customer downgrade a pricing plan?\n",
" * For a subscription business model, will a customer renew his/her subscription?\n",
"\n",
"### Machine learning problem formulation\n",
"\n",
"#### Classification: will this customer churn?\n",
"\n",
"To goal of classification is to identify the at-risk customers and sometimes their unusual behavior, such as: will this customer churn or downgrade their plan? Is there any unusual behavior for a customer? The latter question can be formulated as an anomaly detection problem.\n",
"\n",
"#### Time Series: will this customer churn in the next X months? When will this customer churn?\n",
"\n",
"You can further explore your users by formulating the problem as a time series one and detect when will the customer churn.\n",
"\n",
"### Data Requirements\n",
"\n",
"#### Data collection Sources\n",
"\n",
"Some most common data sources used to construct a data set for churn analysis are:\n",
"\n",
"* Customer Relationship Management platform (CRM), \n",
"* engagement and usage data (analytics services), \n",
"* passive feedback (ratings based on your request), and active feedback (customer support request, feedback on social media and review platforms).\n",
"\n",
"#### Construct a Data Set for Churn Analysis\n",
"\n",
"Most raw data collected from the sources mentioned above are huge and often needs a lot of cleaning and pre-processing. For example, usage data is usually event-based log data and can be more than a few gigabytes every day; you can aggregate the data to user-level daily for further analysis. Feedback and review data are mostly text data, so you would need to clean and pre-process the natural language data to be normalized, machine-readable data. If you are joining multiple data sources (especially from different platforms) together, you would want to make sure all data points are consistent, and the user identity can be matched across different platforms.\n",
" \n",
"#### Challenges with Customer Churn\n",
"\n",
"* Business related\n",
" * Importance of domain knowledge: this is critical when you start building features for the machine learning model. It is important to understand the business enough to decide which features would trigger retention.\n",
"* Data issues\n",
" * fewer churn data available (imbalanced classes): data for churn analysis is often very imbalanced as most of the customers of a business are happy customers (usually).\n",
" * User identity mapping problem: if you are joining data from different platforms (CRM, email, feedback, mobile app, and website usage data), you would want to make sure user A is recognized as the same user across multiple platforms. There are third-party solutions that help you tackle this problem.\n",
" * Not collecting the right data for the use case or Lacking enough data"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"## Data Selection\n",
"\n",
"You will use generated music streaming data that is simulated to imitate music streaming user behaviors. The data simulated contains 1100 users and their user behavior for one year (2019/10/28 - 2020/10/28). Data is simulated using the [EventSim](https://github.com/Interana/eventsim) and does not contain any real user data.\n",
"\n",
"* Observation window: you will use 1 year of data to generate predictions.\n",
"* Explanation of fields:\n",
" * `ts`: event UNIX timestamp\n",
" * `userId`: a randomly assigned unique user id\n",
" * `sessionId`: a randomly assigned session id unique to each user\n",
" * `page`: event taken by the user, e.g. \"next song\", \"upgrade\", \"cancel\"\n",
" * `auth`: whether the user is a logged-in user\n",
" * `method`: request method, GET or PUT\n",
" * `status`: request status\n",
" * `level`: if the user is a free or paid user\n",
" * `itemInSession`: event happened in the session\n",
" * `location`: location of the user's IP address\n",
" * `userAgent`: agent of the user's device\n",
" * `lastName`: user's last name\n",
" * `firstName`: user's first name\n",
" * `registration`: user's time of registration\n",
" * `gender`: gender of the user\n",
" * `artist`: artist of the song the user is playing at the event\n",
" * `song`: song title the user is playing at the event\n",
" * `length`: length of the session\n",
" \n",
" \n",
" * the data will be downloaded from Github and contained in an [Amazon Simple Storage Service](https://aws.amazon.com/s3/) (Amazon S3) bucket."
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"For this specific use case, you will focus on a solution to predict whether a customer will cancel the subscription. Some possible expansion of the work includes:\n",
"\n",
"* predict plan downgrading\n",
"* when a user will churn\n",
"* add song attributes (genre, playlist, charts) and user attributes (demographics) to the data\n",
"* add user feedback and customer service requests to the data\n"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"## PART 1: Prepare Data"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"### Set Up Notebook"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"!pip install -q 's3fs==0.4.2' 'sagemaker-experiments'\n",
"!pip install --upgrade sagemaker boto3\n",
"# s3fs is needed for pandas to read files from S3"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"import sagemaker\n",
"import json\n",
"import pandas as pd\n",
"import glob\n",
"import s3fs\n",
"import boto3\n",
"import numpy as np"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"### Parameters \n",
"The following lists configurable parameters that are used throughout the whole notebook."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"sagemaker_session = sagemaker.Session()\n",
"bucket = sagemaker_session.default_bucket() # replace with your own bucket name if you have one\n",
"s3 = sagemaker_session.boto_session.resource(\"s3\")\n",
"\n",
"region = boto3.Session().region_name\n",
"role = sagemaker.get_execution_role()\n",
"smclient = boto3.Session().client(\"sagemaker\")\n",
"\n",
"prefix = \"music-streaming\""
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"### Ingest Data\n",
"\n",
"We ingest the simulated data from the public SageMaker S3 training database."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"##### Alternative: copy data from a public S3 bucket to your own bucket\n",
"##### data file should include full_data.csv and sample.json\n",
"#### cell 5 - 7 is not needed; the processing job before data wrangler screenshots is not needed\n",
"!mkdir -p data/raw\n",
"s3 = boto3.client(\"s3\")\n",
"s3.download_file(\n",
" f\"sagemaker-example-files-prod-{region}\",\n",
" \"datasets/tabular/customer-churn/customer-churn-data-v2.zip\",\n",
" \"data/raw/customer-churn-data.zip\",\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"!unzip -o ./data/raw/customer-churn-data.zip -d ./data"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"# unzip the partitioned data files into the same folder\n",
"!unzip -o data/simu-1.zip -d data/raw\n",
"!unzip -o data/simu-2.zip -d data/raw\n",
"!unzip -o data/simu-3.zip -d data/raw\n",
"!unzip -o data/simu-4.zip -d data/raw"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"!rm ./data/raw/*.zip"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"!unzip -o data/sample.zip -d data/raw"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"!aws s3 cp ./data/raw s3://$bucket/$prefix/data/json/ --recursive"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"### Data Cleaning\n",
"\n",
"Due to the size of the data (~2GB), you will start exploring our data starting with a smaller sample, decide which pre-processing steps are necessary, and apply them to the whole dataset."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"import os\n",
"\n",
"# if your SageMaker Studio notebook's memory is getting full, you can run the following command to remove the raw data files from the instance and free up some memory.\n",
"# You will read data from your S3 bucket onwards and will not need the raw data stored in the instance.\n",
"os.remove(\"data/simu-1.zip\")\n",
"os.remove(\"data/simu-2.zip\")\n",
"os.remove(\"data/simu-3.zip\")\n",
"os.remove(\"data/simu-4.zip\")\n",
"os.remove(\"data/sample.zip\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"sample_file_name = \"./data/raw/sample.json\"\n",
"# s3_sample_file_name = \"data/json/sample.json\"\n",
"# sample_path = \"s3://{}/{}/{}\".format(bucket, prefix, s3_sample_file_name)\n",
"sample = pd.read_json(sample_file_name, lines=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"sample.head(2)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"#### Remove irrelevant columns\n",
"\n",
"From the first look of data, you can notice that columns `lastName`, `firstName`, `method` and `status` are not relevant features. These will be dropped from the data."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"columns_to_remove = [\"method\", \"status\", \"lastName\", \"firstName\"]\n",
"sample = sample.drop(columns=columns_to_remove)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"#### Check for null values\n",
"\n",
"You are going to remove all events without an `userId` assigned since you are predicting which recognized user will churn from our service. In this case, all the rows(events) have a `userId` and `sessionId` assigned, but you will still run this step for the full dataset. For other columns, there are ~3% of data that are missing some demographic information of the users, and ~20% missing the song attributes, which is because the events contain not only playing a song, but also other actions including login and log out, downgrade, cancellation, etc. There are ~3% of users that do not have a registration time, so you will remove these anonymous users from the record."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"print(\"percentage of the value missing in each column is: \")\n",
"sample.isnull().sum() / len(sample)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"sample = sample[~sample[\"userId\"].isnull()]\n",
"sample = sample[~sample[\"registration\"].isnull()]"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"### Data Exploration\n",
"\n",
"Let's take a look at our categorical columns first: `page`, `auth`, `level`, `location`, `userAgent`, `gender`, `artist`, and `song`, and start with looking at unique values for `page`, `auth`, `level`, and `gender` since the other three have many unique values and you will take a different approach."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"cat_columns = [\"page\", \"auth\", \"level\", \"gender\"]\n",
"cat_columns_long = [\"location\", \"userAgent\", \"artist\", \"song\", \"userId\"]\n",
"for col in cat_columns:\n",
" print(\"The unique values in column {} are: {}\".format(col, sample[col].unique()))\n",
"for col in cat_columns_long:\n",
" print(\"There are {} unique values in column {}\".format(sample[col].nunique(), col))"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"#### Key observations from the above information\n",
"\n",
"* There are 101 unique users with 72 unique locations, this information may not be useful as a categorical feature. You can parse this field and only keep State information, but even that will give us 50 unique values in this category, so you can either remove this column or bucket it to a higher level (NY --> Northeast).\n",
"* Artist and song details might not be helpful as categorical features as there are too many categories; you can quantify these to a user level, i.e. how many artists this user has listened to in total, how many songs this user has played in the last week, last month, in 180 days, in 365 days. You can also bring in external data to get song genres and other artist attributes to enrich this feature.\n",
"* In the column `page`, 'Thumbs Down', 'Thumbs Up', 'Add to Playlist', 'Roll Advert','Help', 'Add Friend', 'Downgrade', 'Upgrade', and 'Error' can all be great features to churn analysis. You will aggregate them to user-level later. There is a \"cancellation confirmation\" value that can be used for the churn indicator.\n",
"\n",
"* Let's take a look at the column `userAgent`:\n"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"UserAgent contains little useful information, but if you care about the browser type and mac/windows difference, you can parse the text and extract the information. Sometimes businesses would love to analyze user behavior based on their App version and device type (iOS v.s. Android), so these could be useful information. In this use case, for modeling purpose, we will remove this column. but you can keep it as a filter for data visualization."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"columns_to_remove = [\"location\", \"userAgent\"]\n",
"sample = sample.drop(columns=columns_to_remove)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"Let's take a closer look at the timestamp columns `ts` and `registration`. We can convert the event timestamp `ts` to year, month, week, day, day of the week, and hour of the day. The registration time should be the same for the same user, so we can aggregate this value to user-level and create a time delta column to calculate the time between registration and the newest event."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"sample[\"date\"] = pd.to_datetime(sample[\"ts\"], unit=\"ms\")\n",
"sample[\"ts_year\"] = sample[\"date\"].dt.year\n",
"sample[\"ts_month\"] = sample[\"date\"].dt.month\n",
"sample[\"ts_week\"] = sample[\"date\"].dt.week\n",
"sample[\"ts_day\"] = sample[\"date\"].dt.day\n",
"sample[\"ts_dow\"] = sample[\"date\"].dt.weekday\n",
"sample[\"ts_hour\"] = sample[\"date\"].dt.hour\n",
"sample[\"ts_date_day\"] = sample[\"date\"].dt.date\n",
"sample[\"ts_is_weekday\"] = [1 if x in [0, 1, 2, 3, 4] else 0 for x in sample[\"ts_dow\"]]\n",
"sample[\"registration_ts\"] = pd.to_datetime(sample[\"registration\"], unit=\"ms\").dt.date"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"#### Define Churn\n",
"\n",
"In this use case, you will use `page == \"Cancellation Confirmation\"` as the indicator of a user churn. You can also use `page == 'downgrade` if you are interested in users downgrading their payment plan. There are ~13% users churned, so you will need to up-sample or down-sample the full dataset to deal with the imbalanced class, or carefully choose your algorithms."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"print(\n",
" \"There are {:.2f}% of users churned in this dataset\".format(\n",
" (\n",
" (sample[sample[\"page\"] == \"Cancellation Confirmation\"][\"userId\"].nunique())\n",
" / sample[\"userId\"].nunique()\n",
" )\n",
" * 100\n",
" )\n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"You can label a user by adding a churn label at a event level then aggregate this value to user level. "
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"sample[\"churned_event\"] = [1 if x == \"Cancellation Confirmation\" else 0 for x in sample[\"page\"]]\n",
"sample[\"user_churned\"] = sample.groupby(\"userId\")[\"churned_event\"].transform(\"max\")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"#### Imbalanced Class\n",
"\n",
"Imbalanced class (much more positive cases than negative cases) is very common in churn analysis. It can be misleading for some machine learning model as the accuracy will be biased towards the majority class. Some useful tactics to deal with imbalanced class are [SMOTE](https://imbalanced-learn.readthedocs.io/en/stable/generated/imblearn.over_sampling.SMOTE.html), use algorithms that are less sensitive to imbalanced class like a tree-based algorithm or use a cost-sensitive algorithm that penalizes wrongly classified minority class."
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"To Summarize every pre-processing steps you have covered:\n",
"* null removals\n",
"* drop irrelevant columns\n",
"* convert event timestamps to features used for analysis and modeling: year, month, week, day, day of week, hour, date, if the day is weekday or weekend, and convert registration timestamp to UTC.\n",
"* create labels (whether the user churned eventually), which is calculated by if one churn event happened in the user's history, you can label the user as a churned user (1). "
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"#### Exploring Data\n",
"\n",
"Based on the available data, look at every column, and decide if you can create a feature from it. For all the columns, here are some directions to explore:\n",
"\n",
" * `ts`: distribution of activity time: time of the day, day of the week\n",
" * `sessionId`: average number of sessions per user\n",
" * `page`: number of thumbs up/thumbs down, added to the playlist, ads, add friend, if the user has downgrade or upgrade the plan, how many errors the user has encountered.\n",
" * `level`: if the user is a free or paid user\n",
" * `registration`: days the user being active, time the user joined the service\n",
" * `gender`: gender of the user\n",
" * `artist`: average number of artists the user listened to\n",
" * `song`: average number of songs listened per user\n",
" * `length`: average time spent per day per user\n",
" \n",
"**Activity Time**\n",
"\n",
"1. Weekday v.s. weekend trends for churned users and active users. It seems like churned users are more active on weekdays than weekends whereas active users do not show a strong difference between weekday v.s. weekends. You can create some features from here: for each user, average events per day -- weekends, average events per day -- weekdays. You can also create features - average events per day of the week, but that will be converted to 7 features after one-hot-encoding, which may be less informative than the previous method.\n",
"2. In terms of hours active during a day, our simulated data did not show a significant difference between day and night for both sets of users. You can have it on your checklist for your analysis, and similarly for the day of the month, the month of the year when you have more than a year of data."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"import seaborn as sns\n",
"import matplotlib.pyplot as plt\n",
"\n",
"events_per_day_per_user = (\n",
" sample.groupby([\"userId\", \"ts_date_day\", \"ts_is_weekday\", \"user_churned\"])\n",
" .agg({\"page\": \"count\"})\n",
" .reset_index()\n",
")\n",
"events_dist = (\n",
" events_per_day_per_user.groupby([\"userId\", \"ts_is_weekday\", \"user_churned\"])\n",
" .agg({\"page\": \"mean\"})\n",
" .reset_index()\n",
")\n",
"\n",
"\n",
"def trend_plot(\n",
" df, plot_type, x, y, hue=None, title=None, x_axis=None, y_axis=None, xticks=None, yticks=None\n",
"):\n",
" if plot_type == \"box\":\n",
" fig = sns.boxplot(x=\"page\", y=y, data=df, hue=hue, orient=\"h\")\n",
" elif plot_type == \"bar\":\n",
" fig = sns.barplot(x=x, y=y, data=df, hue=hue)\n",
"\n",
" sns.set(rc={\"figure.figsize\": (12, 3)})\n",
" sns.set_palette(\"Set2\")\n",
" sns.set_style(\"darkgrid\")\n",
" plt.title(title)\n",
" plt.xlabel(x_axis)\n",
" plt.ylabel(y_axis)\n",
" plt.yticks([0, 1], yticks)\n",
" return plt.show(fig)\n",
"\n",
"\n",
"trend_plot(\n",
" events_dist,\n",
" \"box\",\n",
" \"page\",\n",
" \"user_churned\",\n",
" \"ts_is_weekday\",\n",
" \"Weekday V.S. Weekends - Average events per day per user\",\n",
" \"average events per user per day\",\n",
" yticks=[\"active users\", \"churned users\"],\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"events_per_hour_per_user = (\n",
" sample.groupby([\"userId\", \"ts_date_day\", \"ts_hour\", \"user_churned\"])\n",
" .agg({\"page\": \"count\"})\n",
" .reset_index()\n",
")\n",
"events_dist = (\n",
" events_per_hour_per_user.groupby([\"userId\", \"ts_hour\", \"user_churned\"])\n",
" .agg({\"page\": \"mean\"})\n",
" .reset_index()\n",
" .groupby([\"ts_hour\", \"user_churned\"])\n",
" .agg({\"page\": \"mean\"})\n",
" .reset_index()\n",
")\n",
"trend_plot(\n",
" events_dist,\n",
" \"bar\",\n",
" \"ts_hour\",\n",
" \"page\",\n",
" \"user_churned\",\n",
" \"Hourly activity - Average events per hour of day per user\",\n",
" \"hour of the day\",\n",
" \"average events per user per hour\",\n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"**Listening Behavior**\n",
"\n",
"You can look at some basic stats for a user's listening habits. Churned users generally listen to a wider variety of songs and artists and spend more time on the App/be with the App longer.\n",
"* Average total: number of sessions, App usage length, number of songs listened, number of artists listened per user, number of ad days active\n",
"* Average daily: number of sessions, App usage length, number of songs listened, number of artists listened per user\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"stats_per_user = (\n",
" sample.groupby([\"userId\", \"user_churned\"])\n",
" .agg(\n",
" {\n",
" \"sessionId\": \"count\",\n",
" \"song\": \"nunique\",\n",
" \"artist\": \"nunique\",\n",
" \"length\": \"sum\",\n",
" \"ts_date_day\": \"count\",\n",
" }\n",
" )\n",
" .reset_index()\n",
")\n",
"avg_stats_group = (\n",
" stats_per_user.groupby([\"user_churned\"])\n",
" .agg(\n",
" {\n",
" \"sessionId\": \"mean\",\n",
" \"song\": \"mean\",\n",
" \"artist\": \"mean\",\n",
" \"length\": \"mean\",\n",
" \"ts_date_day\": \"mean\",\n",
" }\n",
" )\n",
" .reset_index()\n",
")\n",
"\n",
"print(\n",
" \"Average total: number of sessions, App usage length, number of songs listened, number of artists listened per user, days active: \"\n",
")\n",
"avg_stats_group"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"stats_per_user = (\n",
" sample.groupby([\"userId\", \"ts_date_day\", \"user_churned\"])\n",
" .agg({\"sessionId\": \"count\", \"song\": \"nunique\", \"artist\": \"nunique\", \"length\": \"sum\"})\n",
" .reset_index()\n",
")\n",
"avg_stats_group = (\n",
" stats_per_user.groupby([\"user_churned\"])\n",
" .agg({\"sessionId\": \"mean\", \"song\": \"mean\", \"artist\": \"mean\", \"length\": \"mean\"})\n",
" .reset_index()\n",
")\n",
"print(\n",
" \"Average daily: number of sessions, App usage length, number of songs listened, number of artists listened per user: \"\n",
")\n",
"avg_stats_group"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"**App Usage Behavior**\n",
"\n",
"You can further explore how the users are using the App besides just listening: number of thumbs up/thumbs down, added to playlist, ads, add friend, if the user has downgrade or upgrade the plan, how many errors the user has encountered. Churned users are slightly more active than other users, and also encounter more errors, listened to more ads, and more downgrade and upgrade. These can be numerical features (number of total events per type per user), or more advanced time series numerical features (errors in last 7 days, errors in last month, etc.)."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"events_list = [\n",
" \"NextSong\",\n",
" \"Thumbs Down\",\n",
" \"Thumbs Up\",\n",
" \"Add to Playlist\",\n",
" \"Roll Advert\",\n",
" \"Add Friend\",\n",
" \"Downgrade\",\n",
" \"Upgrade\",\n",
" \"Error\",\n",
"]\n",
"usage_column_name = []\n",
"for event in events_list:\n",
" event_name = \"_\".join(event.split()).lower()\n",
" usage_column_name.append(event_name)\n",
" sample[event_name] = [1 if x == event else 0 for x in sample[\"page\"]]"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"app_use_per_user = sample.groupby([\"userId\", \"user_churned\"])[usage_column_name].sum().reset_index()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"app_use_group = app_use_per_user.groupby([\"user_churned\"])[usage_column_name].mean().reset_index()\n",
"app_use_group"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"## Pre-processing with SageMaker Data Wrangler\n",
"\n",
"Now that you have a good understanding of your data and decided which steps are needed to pre-process your data, you can utilize the new Amazon SageMaker GUI tool **Data Wrangler**, without writing all the code for the SageMaker Processing Job.\n",
"\n",
"* Here we used a Processing Job to convert the raw streaming data files downloaded from the github repo (`simu-*.zip` files) to a full, CSV formatted file for Data Wrangler Ingestion purpose.\n",
"you are importing the raw streaming data files downloaded from the github repo (`simu-*.zip` files). The raw JSON files were converted to CSV format and combined to one file for Data Wrangler Ingestion purpose."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"!pip install -U sagemaker"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"%%writefile preprocessing_predw.py\n",
"\n",
"import argparse\n",
"import os\n",
"import warnings\n",
"import glob\n",
"import time\n",
"import pandas as pd\n",
"import json\n",
"import argparse\n",
"\n",
"from sklearn.exceptions import DataConversionWarning\n",
"\n",
"warnings.filterwarnings(action=\"ignore\", category=DataConversionWarning)\n",
"start_time = time.time()\n",
"\n",
"if __name__ == \"__main__\":\n",
" parser = argparse.ArgumentParser()\n",
" parser.add_argument(\"--processing-output-filename\")\n",
"\n",
" args, _ = parser.parse_known_args()\n",
" print(\"Received arguments {}\".format(args))\n",
"\n",
" input_jsons = glob.glob(\"/opt/ml/processing/input/data/**/*.json\", recursive=True)\n",
"\n",
" df_all = pd.DataFrame()\n",
" for name in input_jsons:\n",
" print(\"\\nStarting file: {}\".format(name))\n",
" df = pd.read_json(name, lines=True)\n",
" df_all = df_all.append(df)\n",
"\n",
" output_filename = args.processing_output_filename\n",
" final_features_output_path = os.path.join(\"/opt/ml/processing/output\", output_filename)\n",
" print(\"Saving processed data to {}\".format(final_features_output_path))\n",
" df_all.to_csv(final_features_output_path, header=True, index=False)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"from sagemaker.sklearn.processing import SKLearnProcessor\n",
"\n",
"sklearn_processor = SKLearnProcessor(\n",
" framework_version=\"1.2-1\", role=role, instance_type=\"ml.m5.xlarge\", instance_count=1\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"s3_client = boto3.client(\"s3\")\n",
"list_response = s3_client.list_objects_v2(Bucket=bucket, Prefix=f\"{prefix}/data/json\")\n",
"s3_input_uris = [f\"s3://{bucket}/{i['Key']}\" for i in list_response[\"Contents\"]]\n",
"s3_input_uris"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
"\n",
"processing_inputs = []\n",
"for i in s3_input_uris:\n",
" name = i.split(\"/\")[-1].split(\".\")[0]\n",
" processing_input = ProcessingInput(\n",
" source=i, input_name=name, destination=f\"/opt/ml/processing/input/data/{name}\"\n",
" )\n",
" processing_inputs.append(processing_input)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"pycharm": {
"name": "#%%\n"
}
},
"outputs": [],
"source": [
"%%time\n",
"processing_output_path = f\"s3://{bucket}/{prefix}/data/processing\"\n",
"final_features_filename = \"full_data.csv\"\n",
"\n",
"sklearn_processor.run(\n",
" code=\"preprocessing_predw.py\",\n",
" inputs=processing_inputs,\n",
" outputs=[\n",
" ProcessingOutput(\n",
" output_name=\"processed_data\",\n",
" source=\"/opt/ml/processing/output\",\n",
" destination=processing_output_path,\n",
" )\n",
" ],\n",
" arguments=[\"--processing-output-filename\", final_features_filename],\n",
")\n",
"\n",
"preprocessing_job_description = sklearn_processor.jobs[-1].describe()"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {
"pycharm": {
"name": "#%% md\n"
}
},
"source": [
"Now you can initiate a Data Wrangler flow. An example flow (`dw_example.flow`) is provided in the github repo. \n",
"\n",
"From the SageMaker Studio launcher page, choose **New data flow**, then choose **import from S3** and select processing_output_filename. \n",
"\n",
"\n",
"
![](\"image/mo1.PNG\")
\n",
"
\n",
"
![](\"image/mo2.PNG\")
\n",
"
\n",
"
![](\"image/mo3.PNG\")
\n",
"\n",
"
\n",
"
![](\"image/mo4.PNG\")
\n",
"\n",
"