ingest_data/sagemaker-keyspaces/SageMaker_Keyspaces_ml_example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "bf23f259",
   "metadata": {},
   "source": [
    "## Train Machine Learning Models using Amazon Keyspaces as a Data Source.  \n",
    "\n",
    "Contributors\n",
    "- `Vadim Lyakhovich (AWS)`\n",
    "- `Ram Pathangi (AWS)`\n",
    "- `Parth Patel (AWS)`\n",
    "\n",
    "\n",
    "*Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved*  \n",
    "[*SPDX-License-Identifier: MIT-0*](https://github.com/aws/mit-0)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9fec995d",
   "metadata": {},
   "source": [
    "### Prerequisites\n",
    "\n",
    "The Notebook execution role must include permissions to access Amazon Keyspaces and Assume the role.\n",
    "\n",
    "*  To access `Amazon Keyspaces` database - use `AmazonKeyspacesReadOnlyAccess` or `AmazonKeyspacesFullAccess` managed policies. Use policy of the least privilege when running in production.  \n",
    "\n",
    "See more at\n",
    "[`AWS Identity and Access Management for Amazon Keyspaces`](https://docs.aws.amazon.com/keyspaces/latest/devguide/security-iam.html).\n",
    "\n",
    "* To assume the role, you need to have [`sts:AssumeRole action`](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html) permissions.\n",
    "    ```bash\n",
    "    {\n",
    "      \"Version\": \"2012-10-17\",  \n",
    "      \"Statement\": [  \n",
    "        {  \n",
    "           \"Action\": [  \n",
    "           \"sts:AssumeRole\"  \n",
    "          ],  \n",
    "          \"Effect\": \"Allow\",  \n",
    "          \"Resource\": \"*\"  \n",
    "        }\n",
    "      ]\n",
    "    }\n",
    "    ```\n",
    "\n",
    "#### Note:\n",
    "Amazon Keyspaces is available in the following [AWS Regions](https://docs.aws.amazon.com/keyspaces/latest/devguide/programmatic.endpoints.html).\n",
    "\n",
    "This notebook was tested with `conda_python3` kernel and should work with Python 3.x."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "698592f9",
   "metadata": {},
   "source": [
    "In this notebook,  \n",
    "\n",
    "1. First, we install Sigv4 driver to connect to Amazon Keyspaces\n",
    "\n",
    "> The Amazon Keyspaces SigV4 authentication plugin for Cassandra client drivers enables you to authenticate calls to Amazon Keyspaces ***using IAM access keys instead of username and password***. To learn more about how the Amazon Keyspaces SigV4 plugin enables [`IAM users, roles, and federated identities`](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html) to authenticate in Amazon Keyspaces API requests, see [`AWS Signature Version 4 process (SigV4)`](https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html)\n",
    "\n",
    "2. Next, we establish a connection to Amazon Keyspaces \n",
    "3. Next, we create a new `Keyspace ***blog_(yyyymmdd)***` and a new `table ***online_retail***`\n",
    "3. Next, we will download retail data about customers.\n",
    "3. Next, we will ingest retail data about customers into Keyspaces.\n",
    "3. Next, we use a notebook available within SageMaker Studio to collect data from Keyspaces database, and prepare data for training using KNN Algorithm. Most of our customers use SageMaker Studio for end to end development of ML Use Cases. They could use this notebook as a base and customize it quickly for their use case. Additionally, they will be able to share this with other collaborators without requiring them to install any additional software. \n",
    "3. Next, we will train the data for clustering.\n",
    "3. After the training is complete, we can view the mapping between customer and their associated cluster.\n",
    "3. And finally, Cleanup Step to drop Keyspaces table to avoid future charges. \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fbfbf162",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Install missing packages and import dependencies\n",
    "\n",
    "# Installing Cassanda SigV4\n",
    "%pip install  cassandra-sigv4\n",
    "\n",
    "# Get Security certificate\n",
    "!curl https://certs.secureserver.net/repository/sf-class2-root.crt -O\n",
    "\n",
    "# Import\n",
    "from sagemaker import get_execution_role\n",
    "from cassandra.cluster import Cluster\n",
    "from ssl import SSLContext, PROTOCOL_TLSv1_2, CERT_REQUIRED\n",
    "from cassandra_sigv4.auth import SigV4AuthProvider\n",
    "import boto3\n",
    "\n",
    "import pandas as pd\n",
    "from pandas import DataFrame\n",
    "\n",
    "import csv\n",
    "from cassandra import ConsistencyLevel\n",
    "from datetime import datetime\n",
    "import time\n",
    "from datetime import timedelta\n",
    "\n",
    "import pandas as pd\n",
    "import datetime as dt\n",
    "import seaborn as sns\n",
    "import matplotlib.pyplot as plt\n",
    "from sklearn.cluster import KMeans\n",
    "from sklearn.preprocessing import MinMaxScaler\n",
    "\n",
    "# Getting credentials from the role\n",
    "client = boto3.client(\"sts\")\n",
    "\n",
    "# Get notebook Role\n",
    "role = get_execution_role()\n",
    "role_info = {\"RoleArn\": role, \"RoleSessionName\": \"session1\"}\n",
    "print(role_info)\n",
    "\n",
    "credentials = client.assume_role(**role_info)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92884fc8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Connect to Cassandra Database from SageMaker Notebook using temporary credentials from the Role.\n",
    "session = boto3.session.Session(\n",
    "    aws_access_key_id=credentials[\"Credentials\"][\"AccessKeyId\"],\n",
    "    aws_secret_access_key=credentials[\"Credentials\"][\"SecretAccessKey\"],\n",
    "    aws_session_token=credentials[\"Credentials\"][\"SessionToken\"],\n",
    ")\n",
    "\n",
    "region_name = session.region_name\n",
    "\n",
    "# Set Context\n",
    "ssl_context = SSLContext(PROTOCOL_TLSv1_2)\n",
    "ssl_context.load_verify_locations(\"sf-class2-root.crt\")\n",
    "ssl_context.verify_mode = CERT_REQUIRED\n",
    "\n",
    "auth_provider = SigV4AuthProvider(session)\n",
    "\n",
    "keyspaces_host = \"cassandra.\" + region_name + \".amazonaws.com\"\n",
    "\n",
    "cluster = Cluster([keyspaces_host], ssl_context=ssl_context, auth_provider=auth_provider, port=9142)\n",
    "session = cluster.connect()\n",
    "\n",
    "\n",
    "# Read data from Keyspaces system table.  Keyspaces is serverless DB so you don't have to create Keyspaces DB ahead of time.\n",
    "r = session.execute(\"select * from system_schema.keyspaces\")\n",
    "\n",
    "# Read Keyspaces row into Panda DataFrame\n",
    "df = DataFrame(r)\n",
    "print(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "835508e5",
   "metadata": {},
   "source": [
    "## Download Sample data \n",
    "For this example we are using public repository at http://archive.ics.uci.edu/ml [1]\n",
    "\n",
    "### References ###\n",
    "`[1] Dua, D. and Graff, C. (2019). UCI Machine Learning Repository [http://archive.ics.uci.edu/ml]. Irvine, CA: University of California, School of Information and Computer Science.`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cafb2242",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Download sample data\n",
    "\n",
    "!aws s3 cp s3://sagemaker-sample-files/datasets/tabular/online_retail/online_retail_II_20k.csv ."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8212936b",
   "metadata": {},
   "source": [
    "## In this step we will create a new ***blog_(yyyymmdd)*** keyspace and ***online_retail*** table\n",
    "In this step we will create a new `Keyspace ***blog_yyymmdd***` and `table ***online_retail***`\n",
    "\n",
    "```\n",
    "CREATE KEYSPACE IF NOT EXISTS blog_yyyymmdd\n",
    "WITH\n",
    "    REPLICATION = {'class': 'SingleRegionStrategy'}\n",
    "\n",
    "\n",
    "CREATE TABLE IF NOT EXISTS online_retail ( \n",
    " invoice\ttext,\n",
    " stock_code\ttext,\n",
    " description\ttext,\n",
    " quantity\tint,\n",
    " invoice_date\tdate,\n",
    " price\tdecimal,\n",
    " customer_id\ttext,\n",
    " country\ttext,\n",
    "   PRIMARY KEY (invoice,stock_code));\n",
    "   \n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4e9a6058",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create Keyspace\n",
    "\n",
    "dt = datetime.now()\n",
    "keyspaces_schema = \"blog_\" + str(dt.year) + str(dt.month) + str(dt.day)\n",
    "\n",
    "createKeyspace = \"\"\"CREATE KEYSPACE IF NOT EXISTS %s\n",
    "WITH\n",
    "    REPLICATION = {'class': 'SingleRegionStrategy'}; \"\"\"\n",
    "cr = session.execute(createKeyspace % keyspaces_schema)\n",
    "time.sleep(5)\n",
    "print(\"Keyspace '\" + keyspaces_schema + \"' created\")\n",
    "\n",
    "# Create Table\n",
    "createTable = \"\"\"CREATE TABLE IF NOT EXISTS %s.online_retail ( \n",
    " invoice text,\n",
    " stock_code text,\n",
    " description text,\n",
    " quantity int,\n",
    " invoice_date date,\n",
    " price decimal,\n",
    " customer_id text,\n",
    " country text,\n",
    "   PRIMARY KEY (invoice,stock_code));\n",
    "\"\"\"\n",
    "cr = session.execute(createTable % keyspaces_schema)\n",
    "time.sleep(20)\n",
    "print(\"Table 'online_retail' created\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "90e2c6a8",
   "metadata": {},
   "source": [
    "## Loading Data\n",
    "Reading Online Retail CSV file and ingesting into Keyspaces table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d423e632",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Populate test data.\n",
    "\n",
    "# csv file name\n",
    "filename = \"online_retail_II_20k.csv\"\n",
    "ROW_LIMIT = 20000\n",
    "PRINT_ROWS = 2000\n",
    "\n",
    "\n",
    "# initializing the titles and rows list\n",
    "fields = []\n",
    "rows = []\n",
    "\n",
    "insert = (\n",
    "    \"INSERT INTO \"\n",
    "    + keyspaces_schema\n",
    "    + '.online_retail (\"invoice\",\"stock_code\",\"description\",\"quantity\",\"invoice_date\",\"price\",\"customer_id\",\"country\") VALUES (?,?,?,?,?,?,?,?);'\n",
    ")\n",
    "\n",
    "prepared = session.prepare(insert)\n",
    "prepared.consistency_level = ConsistencyLevel.LOCAL_QUORUM\n",
    "\n",
    "\n",
    "print(\"Start Loading\", ROW_LIMIT, \"rows into the table at\", datetime.now())\n",
    "start_time = time.monotonic()\n",
    "\n",
    "# reading csv file.\n",
    "with open(filename, \"r\", encoding=\"utf-8-sig\") as csvfile:\n",
    "    # creating a csv reader object\n",
    "    csvreader = csv.reader(csvfile)\n",
    "\n",
    "    # extracting field names through first row\n",
    "    fields = next(csvreader)\n",
    "\n",
    "    # extracting each data row one by one\n",
    "    # print(fields)\n",
    "    for row in csvreader:\n",
    "        try:\n",
    "            if (csvreader.line_num % PRINT_ROWS) == 0:\n",
    "                print(\"Rows so far: %d\" % (csvreader.line_num))\n",
    "                print(datetime.now())\n",
    "\n",
    "            # print(row)\n",
    "            inv_date = datetime.strptime(row[4], \"%m/%d/%y %H:%M\")\n",
    "            # print(inv_date)\n",
    "            r = session.execute(\n",
    "                prepared,\n",
    "                (\n",
    "                    str(row[0]),\n",
    "                    str(row[1]),\n",
    "                    str(row[2]),\n",
    "                    int(row[3]),\n",
    "                    inv_date,\n",
    "                    float(row[5]),\n",
    "                    str(row[6]),\n",
    "                    str(row[7]),\n",
    "                ),\n",
    "            )\n",
    "\n",
    "            if csvreader.line_num >= ROW_LIMIT:\n",
    "                break\n",
    "        except Exception as ex:\n",
    "            print(\"Error for row %d\" % (csvreader.line_num))\n",
    "            print(row)\n",
    "            print(ex)\n",
    "\n",
    "    # get total number of rows\n",
    "    print(\"Total no. of rows: %d\" % (csvreader.line_num))\n",
    "\n",
    "end_time = time.monotonic()\n",
    "print(\"Load time:\", timedelta(seconds=end_time - start_time), \"for\", ROW_LIMIT, \"rows\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f3984022",
   "metadata": {},
   "source": [
    "## ML Code"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f81e86eb",
   "metadata": {},
   "source": [
    "Now that we have data in Keyspace, let's read the data from Keyspace into the data frame. Once you have data into data frame you can perform the data cleanup to make sure it’s ready to train the modal. \n",
    "\n",
    "In this example we will also group the data based on Recency, Frequency and Monetary value to generate RFM Matrix."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "876e4ed3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Prepare Data\n",
    "\n",
    "r = session.execute(\"select * from \" + keyspaces_schema + \".online_retail\")\n",
    "\n",
    "df = DataFrame(r)\n",
    "df.head(100)\n",
    "\n",
    "df.count()\n",
    "df[\"description\"].nunique()\n",
    "df[\"totalprice\"] = df[\"quantity\"] * df[\"price\"]\n",
    "df.groupby(\"invoice\").agg({\"totalprice\": \"sum\"}).head()\n",
    "\n",
    "df.groupby(\"description\").agg({\"price\": \"max\"}).sort_values(\"price\", ascending=False).head()\n",
    "df.sort_values(\"price\", ascending=False).head()\n",
    "df[\"country\"].value_counts().head()\n",
    "df.groupby(\"country\").agg({\"totalprice\": \"sum\"}).sort_values(\"totalprice\", ascending=False).head()\n",
    "\n",
    "returned = df[df[\"invoice\"].str.contains(\"C\", na=False)]\n",
    "returned.sort_values(\"quantity\", ascending=True).head()\n",
    "\n",
    "df.isnull().sum()\n",
    "df.dropna(inplace=True)\n",
    "df.isnull().sum()\n",
    "df.dropna(inplace=True)\n",
    "df.isnull().sum()\n",
    "df.describe([0.05, 0.01, 0.25, 0.50, 0.75, 0.80, 0.90, 0.95, 0.99]).T\n",
    "df.drop(df.loc[df[\"customer_id\"] == \"\"].index, inplace=True)\n",
    "\n",
    "# Recency Metric\n",
    "import datetime as dt\n",
    "\n",
    "today_date = dt.date(2011, 12, 9)\n",
    "df[\"customer_id\"] = df[\"customer_id\"].astype(int)\n",
    "\n",
    "# create get the most recent invoice for each customer\n",
    "temp_df = df.groupby(\"customer_id\").agg({\"invoice_date\": \"max\"})\n",
    "temp_df[\"invoice_date\"] = temp_df[\"invoice_date\"].astype(str)\n",
    "temp_df[\"invoice_date\"] = pd.to_datetime(temp_df[\"invoice_date\"]).dt.date\n",
    "temp_df[\"Recency\"] = (today_date - temp_df[\"invoice_date\"]).dt.days\n",
    "recency_df = temp_df.drop(columns=[\"invoice_date\"])\n",
    "recency_df.head()\n",
    "\n",
    "# Frequency Metric\n",
    "temp_df = df.groupby([\"customer_id\", \"invoice\"]).agg({\"invoice\": \"count\"})\n",
    "freq_df = temp_df.groupby(\"customer_id\").agg({\"invoice\": \"count\"})\n",
    "freq_df.rename(columns={\"invoice\": \"Frequency\"}, inplace=True)\n",
    "\n",
    "# Monetary Metric\n",
    "monetary_df = df.groupby(\"customer_id\").agg({\"totalprice\": \"sum\"})\n",
    "monetary_df.rename(columns={\"totalprice\": \"Monetary\"}, inplace=True)\n",
    "rfm = pd.concat([recency_df, freq_df, monetary_df], axis=1)\n",
    "\n",
    "df = rfm\n",
    "df[\"RecencyScore\"] = pd.qcut(df[\"Recency\"], 5, labels=[5, 4, 3, 2, 1])\n",
    "df[\"FrequencyScore\"] = pd.qcut(df[\"Frequency\"].rank(method=\"first\"), 5, labels=[1, 2, 3, 4, 5])\n",
    "df[\"Monetary\"] = df[\"Monetary\"].astype(int)\n",
    "df[\"MonetaryScore\"] = pd.qcut(df[\"Monetary\"], 5, labels=[1, 2, 3, 4, 5])\n",
    "df[\"RFM_SCORE\"] = (\n",
    "    df[\"RecencyScore\"].astype(str)\n",
    "    + df[\"FrequencyScore\"].astype(str)\n",
    "    + df[\"MonetaryScore\"].astype(str)\n",
    ")\n",
    "seg_map = {\n",
    "    r\"[1-2][1-2]\": \"Hibernating\",\n",
    "    r\"[1-2][3-4]\": \"At Risk\",\n",
    "    r\"[1-2]5\": \"Can't Loose\",\n",
    "    r\"3[1-2]\": \"About to Sleep\",\n",
    "    r\"33\": \"Need Attention\",\n",
    "    r\"[3-4][4-5]\": \"Loyal Customers\",\n",
    "    r\"41\": \"Promising\",\n",
    "    r\"51\": \"New Customers\",\n",
    "    r\"[4-5][2-3]\": \"Potential Loyalists\",\n",
    "    r\"5[4-5]\": \"Champions\",\n",
    "}\n",
    "\n",
    "df[\"Segment\"] = df[\"RecencyScore\"].astype(str) + rfm[\"FrequencyScore\"].astype(str)\n",
    "df[\"Segment\"] = df[\"Segment\"].replace(seg_map, regex=True)\n",
    "df.head()\n",
    "rfm = df.loc[:, \"Recency\":\"Monetary\"]\n",
    "df.groupby(\"customer_id\").agg({\"Segment\": \"sum\"}).head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "435c1499",
   "metadata": {},
   "source": [
    "Now that we have our final dataset, we will start our training."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6715aa6a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Training\n",
    "\n",
    "sc = MinMaxScaler((0, 1))\n",
    "df = sc.fit_transform(rfm)\n",
    "\n",
    "# Clustering\n",
    "kmeans = KMeans(n_clusters=6).fit(df)\n",
    "\n",
    "# Result\n",
    "kumeler = kmeans.labels_"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0e9da533",
   "metadata": {},
   "source": [
    "Let’s visualize the data to see how records are distributed in different clusters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "62dfba86",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Visualize the clusters\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "final_df = pd.DataFrame({\"customer_id\": rfm.index, \"Kumeler\": kumeler})\n",
    "bucket_deta = final_df.groupby(\"Kumeler\").agg({\"customer_id\": \"count\"}).head()\n",
    "index_deta = final_df.groupby(\"Kumeler\").agg({\"Kumeler\": \"max\"}).head()\n",
    "index_deta[\"Kumeler\"] = index_deta[\"Kumeler\"].astype(int)\n",
    "dataFrame = pd.DataFrame(data=bucket_deta[\"customer_id\"], index=index_deta[\"Kumeler\"])\n",
    "dataFrame.rename(columns={\"customer_id\": \"Total Customers\"}).plot.bar(\n",
    "    rot=70, title=\"RFM clustoring\"\n",
    ")\n",
    "# dataFrame.plot.bar(rot=70, title=\"RFM clustoring\");\n",
    "plt.show(block=True);"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "95da7c87",
   "metadata": {},
   "source": [
    "## Cleanup \n",
    "In this step we will drop the Keyspaces to prevent future charges"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6ea31d3",
   "metadata": {},
   "outputs": [],
   "source": [
    "deleteKeyspace = \"DROP KEYSPACE IF EXISTS \" + keyspaces_schema\n",
    "dr = session.execute(deleteKeyspace)\n",
    "time.sleep(5)\n",
    "print(\n",
    "    \"Dropping %s keyspace.  It may take a few seconds to a minute to complete deletion of keyspace and table.\"\n",
    "    % keyspaces_schema\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "87221772",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:081325390199:image/datascience-1.0"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}