process-amazon-reviews.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Process Amazon reviews"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Install dependencie"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install confluent-kafka"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from confluent_kafka import Producer, Consumer, KafkaError # to produce and consume data from Apache Kafka topics\n",
    "import boto3 # to programmatically create, configure, and manage AWS resources\n",
    "import json # to work with social media messages that are represented as JSON objects\n",
    "import re # for helper functionality to clean HTML tags from social media messages\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prepare models"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Use a JumpStart models to predict if a message is positive or negative\n",
    "The model you can use is Text Classification by HuggingFace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a mapping dictionary to map model labels to negative/positive label\n",
    "label_mapping = {'LABEL_0': 'negative', 'LABEL_1': 'positive'}\n",
    "\n",
    "def get_prediction(text):\n",
    "    endpoint_name = 'jumpstart-text-classification' # TODO: Set endpoint name of your model for sentiment analysis\n",
    "    client = boto3.client('runtime.sagemaker')\n",
    "    query_response = client.invoke_endpoint(EndpointName=endpoint_name, ContentType='application/x-text', Body=text, Accept='application/json;verbose')\n",
    "    model_predictions = json.loads(query_response['Body'].read())\n",
    "    probabilities, labels, predicted_label = model_predictions['probabilities'], model_predictions['labels'], model_predictions['predicted_label']\n",
    "    # Map the predicted_label to your the label using the mapping dictionary\n",
    "    predicted_label = label_mapping.get(predicted_label, predicted_label)\n",
    "    return probabilities, labels, predicted_label\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Use a JupmStart zero-shot model to predict categories of the product\n",
    "You can use Zero-Shot Text Classification by HuggingFace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "categories = [\n",
    "  \"Snacks\",\n",
    "  \"Coffee\",\n",
    "  \"Beverages\",\n",
    "  \"Condiments\",\n",
    "  \"Canned Goods\",\n",
    "  \"Bakery and Baked Goods\",\n",
    "  \"Dairy Products\",\n",
    "  \"Sweets and Desserts\",\n",
    "  \"Breakfast Foods\",\n",
    "  \"Grains and Pasta\",\n",
    "  \"Protein Products\",\n",
    "  \"Frozen Foods\",\n",
    "  \"Health and Nutrition\",\n",
    "  \"Baby and Toddler Food\",\n",
    "  \"Pet Supplies\",\n",
    "  \"Fruits and Vegetables\",\n",
    "  \"Herbs and Spices\",\n",
    "  \"Beverage Accessories\",\n",
    "  \"Cooking Oils and Fats\",\n",
    "  \"Ethnic and Specialty Foods\",\n",
    "  \"Dietary Supplements\"\n",
    "];\n",
    "def get_categories(text):\n",
    "    endpoint_name = 'jumpstart-zero-shot' # TODO: Set endpoint name of your model for classification\n",
    "    client = boto3.client('runtime.sagemaker')\n",
    "    \n",
    "    model_input = {\"sequences\": text, \"candidate_labels\": categories, \"multi_class\": True}\n",
    "    query_response = client.invoke_endpoint(\n",
    "        EndpointName=endpoint_name,\n",
    "        ContentType='application/json',\n",
    "        Body=json.dumps(model_input),\n",
    "        Accept='application/json;verbose'\n",
    "    )\n",
    "    \n",
    "    model_predictions = json.loads(query_response['Body'].read())\n",
    "    \n",
    "    # Filter categories with probabilities over 60%\n",
    "    filtered_categories = [\n",
    "        label_mapping.get(label, label)\n",
    "        for label, score in zip(model_predictions['labels'], model_predictions['scores'])\n",
    "        if score > 0.6\n",
    "    ]\n",
    "    \n",
    "    return filtered_categories\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test endpoint for classifications"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(get_categories(\"very delicious coffee for my cat who likes to snack on tiramisu\"))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Apache Kafka workflow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define Apache Kafka connection properties"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO: Load Apache Kafka certificates into certificates folder\n",
    "apache_kafka_ssl_config = {\n",
    "    'ssl.ca.location': 'certificates/ca.pem', \n",
    "    'ssl.certificate.location': 'certificates/service.cert',\n",
    "    'ssl.key.location': 'certificates/service.key',\n",
    "    'security.protocol': 'ssl',\n",
    "}\n",
    "\n",
    "apache_kafka_uri = '' # TODO: Set URI for Apache Kafka\n",
    "\n",
    "apache_kafka_input_topic_name = 'reviews'\n",
    "apache_kafka_enriched_output_topic_name = 'reviews_enriched'\n",
    "apache_kafka_processing_errors_topic_name = 'reviews_processing_errors'\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Apache Kafka Consumer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "consumer = Consumer({'bootstrap.servers': apache_kafka_uri, 'group.id': 'mygroup', 'auto.offset.reset': 'earliest', **apache_kafka_ssl_config})\n",
    "consumer.subscribe([apache_kafka_input_topic_name])\n",
    "\n",
    "CLEANR = re.compile('<.*?>') \n",
    "\n",
    "def get_clean_content(json_object):    \n",
    "    content = json_object.get(\"Text\", \"\")  # Retrieve 'content' property    \n",
    "    only_text = re.sub(CLEANR, '', content)\n",
    "    return only_text\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Apache Kafka Producer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "producer = Producer({\n",
    "    'bootstrap.servers': apache_kafka_uri, \n",
    "    **apache_kafka_ssl_config \n",
    "})\n",
    "\n",
    "# Send a message to a Kafka topic\n",
    "def send_message(message, topic_name):\n",
    "    producer.produce(topic_name, json.dumps(message).encode('utf-8'))\n",
    "    producer.flush()\n",
    "    \n",
    "def send_enriched_data(message, probabilities, predicted_label, key_categories):\n",
    "    message['probabilities'] = probabilities\n",
    "    message['sentiment_predition'] = predicted_label\n",
    "    message['categories'] = key_categories\n",
    "    send_message(message, apache_kafka_enriched_output_topic_name)\n",
    "    \n",
    "def report_processing_error(message, error_code, error_message):\n",
    "    message['processing_error_code'] = error_code\n",
    "    message['processing_error_message'] = error_message\n",
    "    send_message(message, apache_kafka_processing_errors_topic_name)\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read messages from Apache Kafka **input topic** and push processed data back to **output topic**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Processing messages\")\n",
    "while True:\n",
    "    message = consumer.poll(1.0)  # Poll for messages, with a timeout of 1 second\n",
    "\n",
    "    if message is None:\n",
    "        continue\n",
    "\n",
    "    if message.error():\n",
    "        if message.error().code() == KafkaError._PARTITION_EOF:\n",
    "            # End of partition event\n",
    "            print(f\"Reached end of partition for topic {message.topic()} [{message.partition()}]\")\n",
    "        else:\n",
    "            print(f\"Error while consuming message: {message.error()}\")\n",
    "    else:\n",
    "        decoded = message.value().decode('utf-8')\n",
    "        json_obj = json.loads(decoded)\n",
    "\n",
    "        content_property = get_clean_content(json_obj)\n",
    "        print(f\"Process the messagee: {content_property}\")\n",
    "        if content_property == \"\":\n",
    "            continue\n",
    "\n",
    "        try:\n",
    "            probabilities, labels, predicted_label = get_prediction(content_property)\n",
    "            key_categories = get_categories(content_property)\n",
    "            print(f\"Inference:\\n\"\n",
    "                  f\"Input text: '{content_property}'\\n\"\n",
    "                  f\"Model prediction: {probabilities}\\n\"\n",
    "                  f\"Predicted label: {predicted_label}\\n\"\n",
    "                  f\"Predicted key_categories: {key_categories}\\n\")\n",
    "\n",
    "            send_enriched_data(json_obj, probabilities, predicted_label, key_categories)\n",
    "            \n",
    "\n",
    "        except Exception as e:\n",
    "            print(f\"An error occurred: {e}\")\n",
    "            response = getattr(e, \"response\", {})\n",
    "            error_code = response.get(\"Error\", {}).get(\"Code\", \"Unknown\")\n",
    "            error_message = response.get(\"Error\", {}).get(\"Message\", \"Unknown\")\n",
    "            report_processing_error(json_obj, error_code, error_message)\n",
    "            \n",
    "\n",
    "# Close the consumer\n",
    "consumer.close()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "language": "python",
   "name": "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.10.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}