Kafka_producer(Step 4).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "9a3edfa6-6e02-420e-bdab-0497f4d3118e",
   "metadata": {},
   "source": [
    "# Kafka Producer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "30bee6bb-ddf2-4c36-8dee-d12e7b861b45",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import joblib\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4f015213-c0e5-4617-92be-2f10764f8c79",
   "metadata": {},
   "source": [
    "# Loading the Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "b09bf427-165b-4afa-9d5b-054aad3c8b5b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "loaded_model=joblib.load('../Model/model_4')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d0d515c2-eb0b-4ab2-aa1c-e46c4fd152a6",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df=pd.read_csv('third_result.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "b35ea56b-e341-4d21-9ca2-f495bba98f2c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "x=df.drop(['Date','Time','Anomaly_Score','Anomaly'],axis=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "41fe4206-aa48-43e5-9e3f-09bccb6732f7",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Host</th>\n",
       "      <th>Method</th>\n",
       "      <th>Endpoint</th>\n",
       "      <th>Protocol</th>\n",
       "      <th>Status Code</th>\n",
       "      <th>Content Size</th>\n",
       "      <th>No of Requests</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2415145644</td>\n",
       "      <td>4</td>\n",
       "      <td>236</td>\n",
       "      <td>6</td>\n",
       "      <td>6</td>\n",
       "      <td>248.0</td>\n",
       "      <td>62</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2757421940</td>\n",
       "      <td>4</td>\n",
       "      <td>10</td>\n",
       "      <td>6</td>\n",
       "      <td>0</td>\n",
       "      <td>5952.0</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1123632510</td>\n",
       "      <td>4</td>\n",
       "      <td>400</td>\n",
       "      <td>6</td>\n",
       "      <td>7</td>\n",
       "      <td>146.0</td>\n",
       "      <td>20</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1123632510</td>\n",
       "      <td>4</td>\n",
       "      <td>177</td>\n",
       "      <td>6</td>\n",
       "      <td>0</td>\n",
       "      <td>491.0</td>\n",
       "      <td>20</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>3118324652</td>\n",
       "      <td>4</td>\n",
       "      <td>40</td>\n",
       "      <td>6</td>\n",
       "      <td>7</td>\n",
       "      <td>548.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5731</th>\n",
       "      <td>2584535982</td>\n",
       "      <td>7</td>\n",
       "      <td>240</td>\n",
       "      <td>6</td>\n",
       "      <td>7</td>\n",
       "      <td>146.0</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5732</th>\n",
       "      <td>2584535982</td>\n",
       "      <td>4</td>\n",
       "      <td>240</td>\n",
       "      <td>6</td>\n",
       "      <td>7</td>\n",
       "      <td>146.0</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5733</th>\n",
       "      <td>2584535982</td>\n",
       "      <td>7</td>\n",
       "      <td>10</td>\n",
       "      <td>6</td>\n",
       "      <td>8</td>\n",
       "      <td>150.0</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5734</th>\n",
       "      <td>2584535982</td>\n",
       "      <td>7</td>\n",
       "      <td>240</td>\n",
       "      <td>6</td>\n",
       "      <td>7</td>\n",
       "      <td>146.0</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5735</th>\n",
       "      <td>3164952431</td>\n",
       "      <td>4</td>\n",
       "      <td>10</td>\n",
       "      <td>6</td>\n",
       "      <td>0</td>\n",
       "      <td>5952.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5736 rows × 7 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "            Host  Method  Endpoint  Protocol  Status Code  Content Size  \\\n",
       "0     2415145644       4       236         6            6         248.0   \n",
       "1     2757421940       4        10         6            0        5952.0   \n",
       "2     1123632510       4       400         6            7         146.0   \n",
       "3     1123632510       4       177         6            0         491.0   \n",
       "4     3118324652       4        40         6            7         548.0   \n",
       "...          ...     ...       ...       ...          ...           ...   \n",
       "5731  2584535982       7       240         6            7         146.0   \n",
       "5732  2584535982       4       240         6            7         146.0   \n",
       "5733  2584535982       7        10         6            8         150.0   \n",
       "5734  2584535982       7       240         6            7         146.0   \n",
       "5735  3164952431       4        10         6            0        5952.0   \n",
       "\n",
       "      No of Requests  \n",
       "0                 62  \n",
       "1                  5  \n",
       "2                 20  \n",
       "3                 20  \n",
       "4                  1  \n",
       "...              ...  \n",
       "5731               6  \n",
       "5732               6  \n",
       "5733               6  \n",
       "5734               6  \n",
       "5735               1  \n",
       "\n",
       "[5736 rows x 7 columns]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "87683189-43d9-4bdd-91b7-798c5688b769",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "y=df.Anomaly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "0f3e42d1-392a-489a-a40d-138f1ef631d9",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0       0\n",
       "1       0\n",
       "2       0\n",
       "3       0\n",
       "4       0\n",
       "       ..\n",
       "5731    0\n",
       "5732    0\n",
       "5733    0\n",
       "5734    0\n",
       "5735    0\n",
       "Name: Anomaly, Length: 5736, dtype: int64"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ad5d9d4f-dbb5-4313-911f-358dcf3a3ad9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "f9e011f1-55d9-42ac-ba35-5c8d4fc8fe1d",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Connecting to the server and Creating topic for Kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "64fcae1f-db19-4460-88f2-f4fb3f5d8e51",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import json\n",
    "from kafka.admin import NewTopic,KafkaAdminClient\n",
    "from kafka import KafkaProducer\n",
    "from time import sleep"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "869e4367-8581-4745-af18-8a499f74e0bc",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "topic=NewTopic(name='Topic3',num_partitions=3,replication_factor=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "cadd6457-351e-4f45-8cc0-356bdf8403b9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "admin=KafkaAdminClient(bootstrap_servers='localhost:9092')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "01bfcd93-4da2-4bad-bbc9-ff3b1d5180c1",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "ename": "TopicAlreadyExistsError",
     "evalue": "[Error 36] TopicAlreadyExistsError: Request 'CreateTopicsRequest_v3(create_topic_requests=[(topic='Topic3', num_partitions=3, replication_factor=1, replica_assignment=[], configs=[])], timeout=30000, validate_only=False)' failed with response 'CreateTopicsResponse_v3(throttle_time_ms=0, topic_errors=[(topic='Topic3', error_code=36, error_message=\"Topic 'Topic3' already exists.\")])'.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mTopicAlreadyExistsError\u001b[0m                   Traceback (most recent call last)",
      "Cell \u001b[0;32mIn[11], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[43madmin\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcreate_topics\u001b[49m\u001b[43m(\u001b[49m\u001b[43m[\u001b[49m\u001b[43mtopic\u001b[49m\u001b[43m]\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/Project/pro_venv/lib/python3.10/site-packages/kafka/admin/client.py:461\u001b[0m, in \u001b[0;36mKafkaAdminClient.create_topics\u001b[0;34m(self, new_topics, timeout_ms, validate_only)\u001b[0m\n\u001b[1;32m    456\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mNotImplementedError\u001b[39;00m(\n\u001b[1;32m    457\u001b[0m         \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mSupport for CreateTopics v\u001b[39m\u001b[38;5;132;01m{}\u001b[39;00m\u001b[38;5;124m has not yet been added to KafkaAdminClient.\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m    458\u001b[0m         \u001b[38;5;241m.\u001b[39mformat(version))\n\u001b[1;32m    459\u001b[0m \u001b[38;5;66;03m# TODO convert structs to a more pythonic interface\u001b[39;00m\n\u001b[1;32m    460\u001b[0m \u001b[38;5;66;03m# TODO raise exceptions if errors\u001b[39;00m\n\u001b[0;32m--> 461\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_send_request_to_controller\u001b[49m\u001b[43m(\u001b[49m\u001b[43mrequest\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/Project/pro_venv/lib/python3.10/site-packages/kafka/admin/client.py:407\u001b[0m, in \u001b[0;36mKafkaAdminClient._send_request_to_controller\u001b[0;34m(self, request)\u001b[0m\n\u001b[1;32m    405\u001b[0m         \u001b[38;5;28;01mbreak\u001b[39;00m\n\u001b[1;32m    406\u001b[0m     \u001b[38;5;28;01melif\u001b[39;00m error_type \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m Errors\u001b[38;5;241m.\u001b[39mNoError:\n\u001b[0;32m--> 407\u001b[0m         \u001b[38;5;28;01mraise\u001b[39;00m error_type(\n\u001b[1;32m    408\u001b[0m             \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mRequest \u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;132;01m{}\u001b[39;00m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m failed with response \u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;132;01m{}\u001b[39;00m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m.\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m    409\u001b[0m             \u001b[38;5;241m.\u001b[39mformat(request, response))\n\u001b[1;32m    410\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m    411\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m response\n",
      "\u001b[0;31mTopicAlreadyExistsError\u001b[0m: [Error 36] TopicAlreadyExistsError: Request 'CreateTopicsRequest_v3(create_topic_requests=[(topic='Topic3', num_partitions=3, replication_factor=1, replica_assignment=[], configs=[])], timeout=30000, validate_only=False)' failed with response 'CreateTopicsResponse_v3(throttle_time_ms=0, topic_errors=[(topic='Topic3', error_code=36, error_message=\"Topic 'Topic3' already exists.\")])'."
     ]
    }
   ],
   "source": [
    "admin.create_topics([topic])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "b51a05e8-38dc-4f74-8b8a-bd08b795cbc3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def json_serializer(data):\n",
    "    return json.dumps(data).encode('utf-8')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "0ae37226-493f-4237-84b1-d382be316cda",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "producer=KafkaProducer(bootstrap_servers='localhost:9092',value_serializer=json_serializer)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "e3c5e18b-5d3c-4d49-b152-478b4acc50cc",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'error_code': 17,\n",
       "  'topic': '<kafka.admin.new_topic.NewTopic object at 0x7f3bf3c821d0>',\n",
       "  'is_internal': False,\n",
       "  'partitions': []}]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "admin.describe_topics([topic])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e90c7842-13f7-49b0-8934-5e6076438a7c",
   "metadata": {},
   "source": [
    "# Sending the Data to the Consumer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "b2afe50b-8699-48a4-ba55-63696bef1c49",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "for index,row in df.iterrows():\n",
    "    producer.send(\"topic3\", row.to_json())\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a8052c1a-d4b3-442e-9cc0-00d295e0f673",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c889deab-5517-474e-9f47-a453fde10575",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b353ae3b-64d1-4831-9ef9-6b376641ed15",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}