02b_Importing_Data_(Python_SDK).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Importing Data (with the AWS Python SDK)\n",
    "\n",
    "In this notebook, we'll create our project workspace in Amazon Personalize and import the prepared data - using [Boto3, the AWS SDK for Python](https://boto3.amazonaws.com/v1/documentation/api/latest/index.html).\n",
    "\n",
    "> For an **alternative** approach to the same steps through the [Amazon Personalize console UI](https://console.aws.amazon.com/personalize/home) - see Notebook [02a_Importing_Data_(Console).ipynb](02a_Importing_Data_(Console).ipynb) instead.\n",
    "\n",
    "Before we start, we'll here:\n",
    "\n",
    "- Import the libraries this notebook will use\n",
    "- Load the variables saved from previous steps\n",
    "- Connect to the relevant AWS services as we have before for IAM and S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Python Built-Ins:\n",
    "import json\n",
    "\n",
    "# External Dependencies:\n",
    "import boto3  # AWS SDK for Python\n",
    "\n",
    "# Local Dependencies:\n",
    "import util  # Small tool to print progress spinner\n",
    "\n",
    "# Reload saved variables:\n",
    "%store -r\n",
    "\n",
    "# Connect to AWS services:\n",
    "personalize = boto3.client(\"personalize\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a Dataset Group (DSG)\n",
    "\n",
    "You can think of the **dataset group** as your **project workspace**: It's the container within which your datasets, models and deployments/inferences will be created.\n",
    "\n",
    "A dataset group can contain multiple solutions (models) and campaigns (deployments), but **only one instance of each dataset type** (interactions, items, and users), so:\n",
    "\n",
    "- You can experiment with different *algorithms and models* **within** one dataset group - which we'll do in this example... But\n",
    "- For comparing results with *different datasets/schemas*, you'll usually need to work with **multiple** dataset groups.\n",
    "\n",
    "Since all these steps can be performed through the SDKs/API too, it's absolutely possible to automate pipelines for setting up multiple dataset groups and experiments within them. We'd recommend referring to the MLOps samples in the [official Amazon Personalize samples repository](https://github.com/aws-samples/amazon-personalize-samples) for examples on how to do this.\n",
    "\n",
    "Since we'll experiment only with different model configurations, we'll create a single dataset group in this example."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "create_dsg_response = personalize.create_dataset_group(\n",
    "    name=\"personalize-poc-lab\"\n",
    ")\n",
    "\n",
    "dataset_group_arn = create_dsg_response[\"datasetGroupArn\"]\n",
    "%store dataset_group_arn\n",
    "print(json.dumps(create_dsg_response, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Although creating a dataset group is usually quick, the above call is asynchronous and we need to check the dataset group reaches `ACTIVE` status before moving on.\n",
    "\n",
    "The cell below polls the status to wait until our DSG is ready:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def is_dsg_ready(desc):\n",
    "    status = desc[\"datasetGroup\"][\"status\"]\n",
    "    if status == \"ACTIVE\":\n",
    "        return True\n",
    "    elif \"FAILED\" in status:\n",
    "        raise ValueError(f\"Failed to create Dataset Group!\\n{desc}\")\n",
    "\n",
    "util.progress.polling_spinner(\n",
    "    fn_poll_result=lambda: personalize.describe_dataset_group(datasetGroupArn=dataset_group_arn),\n",
    "    fn_is_finished=is_dsg_ready,\n",
    "    fn_stringify_result=lambda d: d[\"datasetGroup\"][\"status\"],\n",
    "    poll_secs=20,\n",
    "    timeout_secs=20*60,  # Max 20 mins\n",
    ")\n",
    "print(\"Dataset Group Ready\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining Interactions Dataset Schema\n",
    "\n",
    "In this step, we'll need to **define the structure** of our interactions CSV using a JSON schema language - referring to the [Datasets and Schemas](https://docs.aws.amazon.com/personalize/latest/dg/how-it-works-dataset-schema.html) section of the developer guide.\n",
    "\n",
    "> ⚠️ **NOTE** that:\n",
    ">\n",
    "> - The columns list in the JSON must **exactly match** the data file, **including the order of columns**\n",
    "> - Any fields with **missing values** *must* include `null` in their `type` entry to be correctly treated by the model\n",
    "> - Watch out also for the `categorical` attribute, which must be set on string fields where appropriate\n",
    "\n",
    "A comprehensive example schema is provided on the [Interactions Dataset doc page](https://docs.aws.amazon.com/personalize/latest/dg/interactions-datasets.html) for reference.\n",
    "\n",
    "▶️ **CHECK** the schema below matches the `interactions.csv` we created earlier, before running the cell!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "interactions_schema = {\n",
    "    \"type\": \"record\",\n",
    "    \"name\": \"Interactions\",\n",
    "    \"namespace\": \"com.amazonaws.personalize.schema\",\n",
    "    \"fields\": [\n",
    "        {\n",
    "            \"name\": \"USER_ID\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"ITEM_ID\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"EVENT_VALUE\",\n",
    "            \"type\": [\"float\", \"null\"]\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"TIMESTAMP\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"EVENT_TYPE\",\n",
    "            \"type\": \"string\"\n",
    "        }\n",
    "    ],\n",
    "    \"version\": \"1.0\"\n",
    "}\n",
    "\n",
    "create_interactions_schema_resp = personalize.create_schema(\n",
    "    name=\"personalize-movielens-interactions-schema\",\n",
    "    schema=json.dumps(interactions_schema),\n",
    ")\n",
    "\n",
    "interactions_schema_arn = create_interactions_schema_resp[\"schemaArn\"]\n",
    "print(json.dumps(create_interactions_schema_resp, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With the schema created, we can now create our dataset object.\n",
    "\n",
    "Note that this step does not *load* the data yet - just associates the schema to our dataset group:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset_type = \"INTERACTIONS\"\n",
    "create_interactions_ds_resp = personalize.create_dataset(\n",
    "    name=\"personalize-movielens-interactions\",\n",
    "    datasetType=dataset_type,\n",
    "    datasetGroupArn=dataset_group_arn,\n",
    "    schemaArn=interactions_schema_arn,\n",
    ")\n",
    "\n",
    "interactions_dataset_arn = create_interactions_ds_resp[\"datasetArn\"]\n",
    "print(json.dumps(create_interactions_ds_resp, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Importing the Interactions Data\n",
    "\n",
    "In this step, we'll create a **dataset import job** to read our interactions data from S3, validate and load it into our Amazon Personalize dataset group."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "create_interactions_dsimport_resp = personalize.create_dataset_import_job(\n",
    "    jobName=\"personalize-movielens-interactions-01\",\n",
    "    datasetArn=interactions_dataset_arn,\n",
    "    dataSource={\n",
    "        \"dataLocation\": interactions_s3uri,\n",
    "    },\n",
    "    roleArn=personalize_role_arn,\n",
    ")\n",
    "\n",
    "interactions_import_job_arn = create_interactions_dsimport_resp[\"datasetImportJobArn\"]\n",
    "print(json.dumps(create_interactions_dsimport_resp, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> ⏰ Importing the data can take some time - which for our small datasets like our movielens-100k extract is often dominated by **overheads** of starting up and shutting down required processing infrastructure and jobs... Rather than scaling with the number of records.\n",
    ">\n",
    "> For this small sample, the import should take something like 15 minutes.\n",
    "\n",
    "We'll start off the next (item metadata) import job in parallel, and wait for both to complete in a later section.\n",
    "\n",
    "Note that:\n",
    "\n",
    "- We **cannot use the dataset until the import job is complete** (e.g. in `ACTIVE` status)... But also,\n",
    "- **Only the interactions dataset is mandatory** - so it would be possible to skip over the other remaining sections below and start building solutions, as soon as this import job completes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining Item Metadata Schema\n",
    "\n",
    "We'll follow the same general steps for our item metadata set as for the core interactions dataset.\n",
    "\n",
    "First, create a dataset schema using the [Datasets and Schemas](https://docs.aws.amazon.com/personalize/latest/dg/how-it-works-dataset-schema.html) docs and the [sample items schema](https://docs.aws.amazon.com/personalize/latest/dg/items-datasets.html#schema-examples-items) as a guide.\n",
    "\n",
    "▶️ **CHECK** the schema below exactly matches the `item-meta.csv` we created earlier, before running the cell!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "items_schema = {\n",
    "    \"type\": \"record\",\n",
    "    \"name\": \"Items\",\n",
    "    \"namespace\": \"com.amazonaws.personalize.schema\",\n",
    "    \"fields\": [\n",
    "        {\n",
    "            \"name\": \"ITEM_ID\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"GENRES\",\n",
    "            \"type\": \"string\",\n",
    "            # Any string not in the mandatory fields is 'categorical'\n",
    "            \"categorical\": True\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"YEAR\",\n",
    "            # Remember, our year field has some missing values!\n",
    "            \"type\": [\"int\", \"null\"]\n",
    "        }\n",
    "    ],\n",
    "    \"version\": \"1.0\"\n",
    "}\n",
    "\n",
    "create_items_schema_resp = personalize.create_schema(\n",
    "    name=\"personalize-movielens-items-schema\",\n",
    "    schema=json.dumps(items_schema),\n",
    ")\n",
    "\n",
    "items_schema_arn = create_items_schema_resp[\"schemaArn\"]\n",
    "print(json.dumps(create_items_schema_resp, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With the schema created, we can now create our dataset object.\n",
    "\n",
    "Again this step does not *load* the data yet - just associates the schema to our dataset group:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset_type = \"ITEMS\"\n",
    "create_items_ds_resp = personalize.create_dataset(\n",
    "    name=\"personalize-movielens-items\",\n",
    "    datasetType=dataset_type,\n",
    "    datasetGroupArn=dataset_group_arn,\n",
    "    schemaArn=items_schema_arn,\n",
    ")\n",
    "\n",
    "items_dataset_arn = create_items_ds_resp[\"datasetArn\"]\n",
    "print(json.dumps(create_items_ds_resp, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Importing the Items Metadata\n",
    "\n",
    "In this step, we'll create a **dataset import job** to read our item metadata from S3, validate and load it into our Amazon Personalize dataset group."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "create_items_dsimport_resp = personalize.create_dataset_import_job(\n",
    "    jobName=\"personalize-movielens-items-01\",\n",
    "    datasetArn=items_dataset_arn,\n",
    "    dataSource={\n",
    "        \"dataLocation\": items_s3uri,\n",
    "    },\n",
    "    roleArn=personalize_role_arn,\n",
    ")\n",
    "\n",
    "items_import_job_arn = create_items_dsimport_resp[\"datasetImportJobArn\"]\n",
    "print(json.dumps(create_items_dsimport_resp, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> ⏰ As with interactions, importing the data can take some time - which for our small datasets like our movielens-100k extract is often dominated by **overheads** of starting up and shutting down required processing infrastructure and jobs... Rather than scaling with the number of records.\n",
    ">\n",
    "> For this small sample, the import should take something like 15 minutes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Wait for Imports to Complete\n",
    "\n",
    "You can of course check progress of your import jobs through the Personalize console, and also through the API.\n",
    "\n",
    "In the cell below, we set up a simple polling loop to check progress and display updates in the notebook - blocking execution until all jobs are done:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "waiting_arns = [interactions_import_job_arn, items_import_job_arn]\n",
    "\n",
    "def are_imports_finished(descriptions):\n",
    "    for desc in descriptions:\n",
    "        status = desc[\"datasetImportJob\"][\"status\"]\n",
    "        arn = desc[\"datasetImportJob\"][\"datasetImportJobArn\"]\n",
    "        if status == \"ACTIVE\":\n",
    "            waiting_arns.remove(arn)\n",
    "        elif \"FAILED\" in status:\n",
    "            raise ValueError(f\"Data import failed!\\n{desc}\")\n",
    "    if not len(waiting_arns):\n",
    "        return True\n",
    "\n",
    "util.progress.polling_spinner(\n",
    "    fn_poll_result=lambda: map(\n",
    "        lambda arn: personalize.describe_dataset_import_job(datasetImportJobArn=arn),\n",
    "        waiting_arns,\n",
    "    ),\n",
    "    fn_is_finished=are_imports_finished,\n",
    "    fn_stringify_result=lambda d: f\"{len(waiting_arns)} imports in progress\",\n",
    "    poll_secs=30,\n",
    "    timeout_secs=60*60,  # Max 1 hour\n",
    ")\n",
    "print(\"Data imported\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## All set!\n",
    "\n",
    "We've now created our dataset group (project) in Amazon Personalize and imported our source datasets.\n",
    "\n",
    "In the next notebook we'll create and evaluate some recommendation models based on this data:\n",
    "\n",
    "- Follow along in the **AWS Console** with the instructions and screenshots in [03a_Creating_and_Evaluating_Solutions_(Console).ipynb](03a_Creating_and_Evaluating_Solutions_(Console).ipynb), *OR*\n",
    "- Run the same steps in code with the **AWS SDK for Python (Boto3)** by following [03b_Creating_and_Evaluating_Solutions_(Python_SDK).ipynb](03b_Creating_and_Evaluating_Solutions_(Python_SDK).ipynb)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  },
  "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.6.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}