1_producer.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "1021cd8f",
   "metadata": {},
   "source": [
    "<center><h1>Management and Analysis of Physics Dataset (MOD. B) </h1></center>\n",
    "<center><h2> Project 5 - Streaming processing of cosmic rays using Drift Tubes detectors</h2></center>\n",
    "<center><h2>Group 2305</h2></center>\n",
    "\n",
    "<center><style>\n",
    "    table {font-size: 24px;}\n",
    "</style></center>\n",
    "\n",
    "| Last Name        | First Name         |Student ID|\n",
    "|:----------------:|:------------------:|:--------------:|\n",
    "| Bertinelli       | Gabriele           |1219907 (tri)   |\n",
    "| Bhatti           | Roben              |2091187         |\n",
    "| Bonato           | Diego              |2091250         |\n",
    "| Cacciola         | Martina            |2097476         |\n",
    "\n",
    "<left><h2> Part 1 - Producer</h2></left>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "068fde3e",
   "metadata": {},
   "source": [
    "### Import packages and modules "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "7f131aa8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# import kafka\n",
    "from kafka       import KafkaProducer\n",
    "from kafka.admin import KafkaAdminClient, NewTopic\n",
    "\n",
    "\n",
    "import pandas as pd\n",
    "import os\n",
    "import boto3\n",
    "import json\n",
    "import time\n",
    "from tqdm        import tqdm"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b242c3bc",
   "metadata": {},
   "source": [
    "### Producer creation\n",
    "\n",
    "The Producer (Sender) is the Kafka abstraction for publishing data into a Kafka topic.  \n",
    "\n",
    "In order for a Producer to publish a message, we need to specify at least two things:\n",
    "1. The location of the cluster over the network &rarr; `KAFKA_BOOTSTRAP_SERVER`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "f8cb5611",
   "metadata": {},
   "outputs": [],
   "source": [
    "# define the list of brokers in the cluster\n",
    "KAFKA_BOOTSTRAP_SERVER = '##'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "94ba5bfb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# producer definition\n",
    "producer = KafkaProducer(bootstrap_servers = KAFKA_BOOTSTRAP_SERVER,\n",
    "                         batch_size=16000, #16MB\n",
    "                            linger_ms=20  ) #ms a producer is willing to wait before sending a batch out\n",
    "\n",
    "        \n",
    "# KAFKA ADMIN is responsible for creating/deleting topics\n",
    "# connecting to client \n",
    "kafka_admin = KafkaAdminClient(bootstrap_servers = KAFKA_BOOTSTRAP_SERVER)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "9b81be2b",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['data_clean', 'test_clean_1', 'data_raw', '__consumer_offsets']"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kafka_admin.list_topics()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eb2d5ff2",
   "metadata": {},
   "source": [
    "### Topic creation\n",
    "\n",
    "\n",
    "2. The topic to which the messages will be published &rarr; `create_topics`. \n",
    "\n",
    "In addition to the name, it is possible to specify the number of partitions of a topic, `num_partitions`, and the number of replication times of the topic, `replication_factor`.  \n",
    "Kafka Replication Factor refers to the multiple copies of data stored across several Kafka brokers. Since we have just one broker we set `replication_factor=1`.  \n",
    "The number of partitions (for the `data_raw` topic) was varied, along with Spark's parameters, to test the performance of the network in different configurations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "2f398635",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['test_clean_1', '__consumer_offsets']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#delete existing topics\n",
    "\n",
    "if 'data_raw' in kafka_admin.list_topics():\n",
    "    kafka_admin.delete_topics(['data_raw'])\n",
    "    \n",
    "if 'data_clean' in kafka_admin.list_topics():\n",
    "    kafka_admin.delete_topics(['data_clean'])\n",
    "\n",
    "kafka_admin.list_topics()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "59acd701",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['data_clean', 'test_clean_1', 'data_raw', '__consumer_offsets']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data_raw = NewTopic(name='data_raw', \n",
    "                        num_partitions=10,                      \n",
    "                        replication_factor=1)                     #replication factor is 1 (no replication) because we have one broker    \n",
    "\n",
    "data_clean = NewTopic(name='data_clean', \n",
    "                          num_partitions=1, \n",
    "                          replication_factor=1)\n",
    "\n",
    "kafka_admin.create_topics(new_topics=[data_raw, data_clean])\n",
    "kafka_admin.list_topics()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6cc82ea2",
   "metadata": {},
   "source": [
    "### Load data from s3 bucket\n",
    "Key\n",
    "We first connect to the s3 bucket containing tha data. We access each file, through a loop, by the `Key` value and we save it in a Pandas Dataframe. We rename the long-name column to `values` and then we loop in the dataframe and we append each row to a message to send to the `data_raw` topic. When the batch reach the defined `batch_size`, we send the message through `producer.flush()`.\n",
    "\n",
    "The `KafkaProducer.send()` method is asynchronous, which means it enqueues the message on an internal queue. The actual sending of the message to the broker happens later, based on tunable parameters. We use this function to gather the rows inside the batch. \n",
    "To send messages synchronously, we use the `flush()` method of the producer. It ensures that all outstanding messages are sent before proceeding. We use it when the batch is complete and we want to send it to the topic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "a11a14cc",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/kafka/jupy/jupyter_env/lib/python3.8/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'cloud-areapd.pd.infn.it'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    }
   ],
   "source": [
    "# the keys are useless as verify is false\n",
    "s3_client = boto3.client('#',\n",
    "                         endpoint_url='##',\n",
    "                         aws_access_key_id='##',\n",
    "                         aws_secret_access_key='##',\n",
    "                         verify=False)\n",
    "\n",
    "#bucket containing data, the key value is the name of txt file we are going to parse\n",
    "list_obj_contents = s3_client.list_objects(Bucket = '##')['Contents'] \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "a5550539",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/kafka/jupy/jupyter_env/lib/python3.8/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'cloud-areapd.pd.infn.it'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NewFile\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      " 86%|███████████████████████████▍    | 1124999/1310592 [20:26<03:22, 916.99it/s]\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "Cell \u001b[0;32mIn[8], line 43\u001b[0m\n\u001b[1;32m     41\u001b[0m         producer\u001b[38;5;241m.\u001b[39mflush()\n\u001b[1;32m     42\u001b[0m         \u001b[38;5;66;03m# sleep time\u001b[39;00m\n\u001b[0;32m---> 43\u001b[0m         \u001b[43mtime\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msleep\u001b[49m\u001b[43m(\u001b[49m\u001b[43mwait_time\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m     44\u001b[0m         batch_counter\u001b[38;5;241m=\u001b[39m\u001b[38;5;241m0\u001b[39m\n\u001b[1;32m     46\u001b[0m \u001b[38;5;66;03m# send last batch \u001b[39;00m\n",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "batch_counter = 0 \n",
    "\n",
    "\n",
    "### Change these parameters to adjust number of records/s\n",
    "\n",
    "wait_time = 0.95  # we send 1 batch of data every wait_time seconds\n",
    "\n",
    "batch_size = 1000  #number of rows sent for batch \n",
    "\n",
    "\n",
    "### we send more than one batch per second by design, we could also send 1 batch per second\n",
    "\n",
    "\n",
    "### loop into s3 bucket and send batches to the broker simulating a streaming data\n",
    "\n",
    "for obj in list_obj_contents:\n",
    "    \n",
    "    #load each txt file into pandas dataframe\n",
    "    \n",
    "    df=pd.read_csv(s3_client.get_object(Bucket='##', Key=obj['Key']).get('Body'), sep=' ')\n",
    "    df.rename(columns = {'HEAD,FPGA,TDC_CHANNEL,ORBIT_CNT,BX_COUNTER,TDC_MEAS':'values'}, inplace = True)\n",
    "    \n",
    "    #df['string'] = df[df.columns].astype(str).apply(lambda x: ', '.join(x), axis = 1)\n",
    "\n",
    "    print('NewFile') \n",
    "    \n",
    "    for i in tqdm(range(len(df))):\n",
    "        \n",
    "        row = df['values'].iloc[i].encode()\n",
    "        \n",
    "        #row_bytes = bytes(row.to_csv(lineterminator=',', header=False,index=False), encoding='utf-8')\n",
    "        \n",
    "        # append a record to the msg to send\n",
    "        producer.send('data_raw', row)\n",
    "        \n",
    "        #batch counter increaser\n",
    "        batch_counter+=1\n",
    "        \n",
    "        #send message to the topic when we reach batch size \n",
    "        if batch_counter==batch_size:\n",
    "            producer.flush()\n",
    "            # sleep time\n",
    "            time.sleep(wait_time)\n",
    "            batch_counter=0\n",
    "            \n",
    "    # send last batch \n",
    "    producer.flush()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1636b95f",
   "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.8.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}