IR-Complaint-Feedback-Mgmt-System.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#This section for Windows users only\n",
    "\n",
    "import os\n",
    "import sys\n",
    "\n",
    "os.chdir(r\"C:\\spark\\spark-files\")\n",
    "os.curdir\n",
    "if 'SPARK_HOME' not in os.environ:\n",
    "    os.environ['SPARK_HOME'] = 'C:\\spark'\n",
    "    \n",
    "SPARK_HOME = os.environ['SPARK_HOME']\n",
    "\n",
    "sys.path.insert(0,os.path.join(SPARK_HOME,\"python\"))\n",
    "sys.path.insert(0,os.path.join(SPARK_HOME,\"python\",\"lib\"))\n",
    "sys.path.insert(0,os.path.join(SPARK_HOME,\"python\",\"lib\",\"pyspark.zip\"))\n",
    "sys.path.insert(0,os.path.join(SPARK_HOME,\"python\",\"lib\",\"py4j-0.10.4-src.zip\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Importing dependencies\n",
    "\n",
    "from pyspark import SparkContext\n",
    "from pyspark import SparkConf\n",
    "from pyspark.mllib.feature import HashingTF\n",
    "from pyspark.mllib.feature import IDF\n",
    "import operator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Setting up spark context, run it only once in a session\n",
    "\n",
    "conf=SparkConf()\n",
    "conf.set(\"spark.executor.memory\", \"1g\")\n",
    "conf.set(\"spark.cores.max\", \"2\")\n",
    "conf.setAppName(\"IRApp\")\n",
    "sc = SparkContext('local', conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['emergency,unable to sit due to sticky stains on berth. PNR no 4512791357. do help immediately']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Loading & extracting data\n",
    "\n",
    "tweetData = sc.textFile(\"tweets_formatted_data.csv\") #Located in C:\\spark\\spark-files\n",
    "fields = tweetData.map(lambda x: x.split(\",\"))\n",
    "documents = fields.map(lambda x: x[1].lower().split(\" \"))\n",
    "#fields.take(1)\n",
    "tweetData.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "('emergency,unable to sit due to sticky stains on berth. PNR no 4512791357. do help immediately',\n",
       " SparseVector(100000, {916: 4.0775, 1462: 5.687, 16079: 2.253, 16470: 3.6721, 37495: 2.9144, 39740: 2.3911, 52613: 5.687, 59838: 4.3007, 60365: 3.4146, 66215: 4.9938, 75088: 5.687, 76603: 5.687, 88349: 5.687, 95803: 4.9938}))"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Calculatig TF-IDF\n",
    "\n",
    "documentNames = fields.map(lambda x: x[0])\n",
    "hashingTF = HashingTF(100000)\n",
    "article_hash_value = hashingTF.transform(documents)\n",
    "article_hash_value.cache()\n",
    "\n",
    "idf = IDF().fit(article_hash_value)\n",
    "tfidf = idf.transform(article_hash_value)\n",
    "\n",
    "xformedData=tweetData.zip(tfidf)\n",
    "xformedData.cache()\n",
    "xformedData.collect()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "LabeledPoint(1.0, (100000,[916,1462,16079,16470,37495,39740,52613,59838,60365,66215,75088,76603,88349,95803],[4.07753744391,5.68697535634,2.25298815185,3.6720723358,2.9143866341,2.39113849034,5.68697535634,4.30068099522,3.41458740488,4.99382817578,5.68697535634,5.68697535634,5.68697535634,4.99382817578]))"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Processing\n",
    "\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "def convertToLabeledPoint(inVal) :\n",
    "    origAttr=inVal[0].split(\",\")\n",
    "    sentiment = 0.0 if origAttr[0] == \"feedback\" else 1.0\n",
    "    return LabeledPoint(sentiment, inVal[1])\n",
    "\n",
    "tweetLp=xformedData.map(convertToLabeledPoint)\n",
    "tweetLp.cache()\n",
    "tweetLp.collect()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0.0, 0.0), (1.0, 1.0), (1.0, 1.0), (0.0, 0.0), (1.0, 1.0)]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Machine Learning - Naive Bayes\n",
    "\n",
    "from pyspark.mllib.classification import NaiveBayes, NaiveBayesModel\n",
    "model = NaiveBayes.train(tweetLp, 1.0)\n",
    "predictionAndLabel = tweetLp.map(lambda p: \\\n",
    "    (float(model.predict(p.features)), float(p.label)))\n",
    "predictionAndLabel.collect()[35:40]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----------+-----+\n",
      "|label|prediction|count|\n",
      "+-----+----------+-----+\n",
      "|  1.0|       1.0|  215|\n",
      "|  0.0|       1.0|   30|\n",
      "|  1.0|       0.0|   36|\n",
      "|  0.0|       0.0|  308|\n",
      "+-----+----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Forming confusion matrix\n",
    "\n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "predDF = sqlContext.createDataFrame(predictionAndLabel.collect(), [\"prediction\",\"label\"])\n",
    "predDF.groupBy(\"label\",\"prediction\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#saving the model\n",
    "\n",
    "#model.save(sc,\"IRModel\")\n",
    "import pickle\n",
    "with open('IRModel', 'wb') as f:\n",
    "    pickle.dump(model, f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#This section classifies real-time tweets and sends data into MySQL Database\n",
    "\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.mllib.classification import  NaiveBayesModel\n",
    "from pyspark.streaming import StreamingContext\n",
    "from pyspark.streaming.kafka import KafkaUtils\n",
    "from pyspark.mllib.feature import HashingTF\n",
    "from pyspark.mllib.feature import IDF\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "\n",
    "import operator\n",
    "import pickle\n",
    "import json\n",
    "import MySQLdb\n",
    "\n",
    "\n",
    "def insert_tweet(tweet,username,pnr,prediction,tweet_id):\n",
    "    query = \"INSERT INTO tweets(tweet,username,pnr,prediction,tweet_id) VALUES ('\"+tweet+\"','\"+username+\"',\"+str(pnr)+\",\"+str(int(prediction))+\",\"+str(tweet_id)+\");\"\n",
    "    try:\n",
    "        conn = MySQLdb.connect(\"localhost\",\"kunwar\",\"\",\"twitter\" )\n",
    "        cursor = conn.cursor()\n",
    "        cursor.execute(query)\n",
    "#         print(\"Database insertion SUCCESSFUL!!\")\n",
    "        conn.commit()\n",
    "    except MySQLdb.Error as e:\n",
    "        print(e)\n",
    "#         print(\"Database insertion unsuccessful!!\")\n",
    "    finally:\n",
    "        conn.close()\n",
    "\n",
    "from pyspark.streaming import StreamingContext\n",
    "conf = SparkConf().setMaster(\"local[2]\").setAppName(\"Streamer\")\n",
    "sc = SparkContext(conf=conf)\n",
    "sc.setLogLevel(\"ERROR\")\n",
    "\n",
    "ssc = StreamingContext(sc, 10)\n",
    "ssc.checkpoint(\"checkpoint\")\n",
    "kstream = KafkaUtils.createDirectStream(\n",
    "ssc, topics = ['twitterstream'], kafkaParams = {\"metadata.broker.list\": 'localhost:9092'})\n",
    "tweets = kstream.map(lambda x: json.loads(x[1]))\n",
    "\n",
    "with open('IRModel', 'rb') as f:\n",
    "    loadedModel = pickle.load(f)\n",
    "\n",
    "bc_model = sc.broadcast(loadedModel)\n",
    "\n",
    "def process_data(data):\n",
    "\n",
    "#         print(\"Processing data ...\")        \n",
    "\n",
    "        if (not data.isEmpty()):\n",
    "            nbModel=bc_model.value\n",
    "            hashingTF = HashingTF(100000)\n",
    "            tf = hashingTF.transform(data.map(lambda x: x[0].encode('utf-8','ignore')))\n",
    "            tf.cache()\n",
    "            idf = IDF(minDocFreq=2).fit(tf)\n",
    "            tfidf = idf.transform(tf)\n",
    "            tfidf.cache()\n",
    "            prediction=nbModel.predict(tfidf)\n",
    "\n",
    "            temp = []\n",
    "            i=0\n",
    "            for p,q,r in data.collect():\n",
    "                temp.append([])\n",
    "                temp[i].append(p.encode('utf-8','ignore'))\n",
    "                temp[i].append(q)\n",
    "                temp[i].append(r)\n",
    "                i+=1\n",
    "            i=0\n",
    "            for p in prediction.collect():\n",
    "                temp[i].append(p)\n",
    "                i+=1\n",
    "\n",
    "            print(temp)\n",
    "            for i in temp:\n",
    "                insert_tweet(str(i[0]),str(i[1]),\"0\",int(i[3]),int(i[2]))\n",
    "        else:\n",
    "#             print(\"Empty RDD !!!\")        \n",
    "            pass\n",
    "\n",
    "txt = tweets.map(lambda x: (x['text'], x['user']['screen_name'], x['id']))\n",
    "txt.foreachRDD(process_data)\n",
    "\n",
    "ssc.start() \n",
    "ssc.awaitTerminationOrTimeout(1000)\n",
    "ssc.stop(stopGraceFully = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.5.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}