Classification_with_Pyspark.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "# ***Classification with Pyspark***\n",
        "\n",
        "### Author: Salma OUARDI\n",
        " Dataset : [StumbleUpon Evergreen Classification](https://www.kaggle.com/c/stumbleupon/data)\n",
        "\n"
      ],
      "metadata": {
        "id": "_jBlHRP_ceGG"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "J_70znI4cG4b",
        "outputId": "103337e9-0b1d-4d83-b87f-3290c6a8df56"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting pyspark\n",
            "  Downloading pyspark-3.2.0.tar.gz (281.3 MB)\n",
            "\u001b[K     |████████████████████████████████| 281.3 MB 39 kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.9.2\n",
            "  Downloading py4j-0.10.9.2-py2.py3-none-any.whl (198 kB)\n",
            "\u001b[K     |████████████████████████████████| 198 kB 58.5 MB/s \n",
            "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.2.0-py2.py3-none-any.whl size=281805911 sha256=5a91e26c70f4ddb906f168b74248c459e854804ce5d36efdff0557603d8bd357\n",
            "  Stored in directory: /root/.cache/pip/wheels/0b/de/d2/9be5d59d7331c6c2a7c1b6d1a4f463ce107332b1ecd4e80718\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.9.2 pyspark-3.2.0\n"
          ]
        }
      ],
      "source": [
        "!pip install pyspark"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark import SparkContext\n",
        "sc=SparkContext()"
      ],
      "metadata": {
        "id": "652hGVFXcM3b"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XzVTGZdoQUNW"
      },
      "source": [
        "# Features"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## **A function to extract the features**\n",
        "\n",
        "*The features we extracted here are all numeric. We excluded the first 4 culumns* **(url, urlid, boilerplate,alchemy_category)**. We also excluded the last column which contains the target **(—1 is evergreen, while 0 is non-evergreen)**."
      ],
      "metadata": {
        "id": "YXmkH7yXd8fK"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def extract_features_dt(record):\n",
        "  mylist=[]\n",
        "  for i in record[4:25]:\n",
        "    if i != '?':\n",
        "      mylist.append(float(i))\n",
        "    else:\n",
        "      mylist.append(float(0))\n",
        "  return np.array(mylist)"
      ],
      "metadata": {
        "id": "jBC9_vNodMgi"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## **A function to extract the target or label**\n",
        "\n",
        "*We extract the last column* **(-1 for evergreen and 0 for non-evergreen)**."
      ],
      "metadata": {
        "id": "8MBFBVL8orCd"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def extract_label(record):\n",
        "  return int(record[-1])"
      ],
      "metadata": {
        "id": "k9bZhuumguXk"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3EwX3jZ3QUNY"
      },
      "source": [
        "## Change each record to LabeledPoint\n"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "\n",
        "\n",
        "\n",
        "> **LabeledPoint** *is a class that represents the features and labels of a data point.*\n",
        "\n"
      ],
      "metadata": {
        "id": "F06a1E_MrE_x"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.mllib.regression import LabeledPoint"
      ],
      "metadata": {
        "id": "df53uJ-_pwh4"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "lines = sc.textFile('train_noheader.tsv')\n",
        "replace = lines.map(lambda row: row.replace(\"\\\"\", \"\"))\n",
        "split = replace.map(lambda row: row.split(\"\\t\"))\n",
        "label = split.map(lambda p: int(p[-1]))\n",
        "data = split.map(lambda r: LabeledPoint(extract_label(r), extract_features_dt(r)))\n",
        "(training, test) = data.randomSplit([0.8,0.2])\n",
        "training.cache()\n",
        "test.cache()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "kWq77fVBg_Pf",
        "outputId": "5ce191e2-4dbe-4918-a90c-efa9d0d8ae53"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "PythonRDD[3] at RDD at PythonRDD.scala:53"
            ]
          },
          "metadata": {},
          "execution_count": 8
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZNIPxMXdQUNb"
      },
      "source": [
        "# Models"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dn2QxcNHQUNc"
      },
      "source": [
        "## train models"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": true,
        "id": "42FxiPERQUNc"
      },
      "outputs": [],
      "source": [
        "from pyspark.mllib.classification import LogisticRegressionWithLBFGS, SVMWithSGD, NaiveBayes\n",
        "from pyspark.mllib.tree import DecisionTree\n",
        "\n",
        "iterations = 10\n",
        "\n",
        "# logistic regression\n",
        "logisticModel = LogisticRegressionWithLBFGS.train(data, iterations)\n",
        "\n",
        "# SVM\n",
        "svmModel = SVMWithSGD.train(data, iterations)\n",
        "\n",
        "# naive bayes\n",
        "naiveBayesModel = NaiveBayes.train(nbData, 1.0) # 1.0 is the smoothing parameter\n",
        "\n",
        "# decision tree\n",
        "decisionTreeModel = DecisionTree.trainClassifier(data, numClasses = 2, categoricalFeaturesInfo = {}, \n",
        "                                                 impurity = 'entropy', maxDepth = 5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7pltfsJ9QUNd"
      },
      "source": [
        "## prediction"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "collapsed": true,
        "id": "n4fRNilIQUNd"
      },
      "outputs": [],
      "source": [
        "lrLabelPrediction = data.map(lambda x: (x.label, logisticModel.predict(x.features)))\n",
        "svmLabelPrediction = data.map(lambda x: (x.label, svmModel.predict(x.features)))\n",
        "nbLabelPrediction = nbData.map(lambda x: (x.label, naiveBayesModel.predict(x.features)))\n",
        "dtPrediction = decisionTreeModel.predict(data.map(lambda x : x.features))\n",
        "dtLabelPrediction = data.map(lambda x: x.label).zip(dtPrediction)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "UGTewzaeQUNe"
      },
      "source": [
        "## evaluation"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "OUnbioBPQUNf"
      },
      "source": [
        "### accuracy"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "scrolled": true,
        "id": "JI4zcYPBQUNf",
        "outputId": "c3a07631-ee0b-449f-b1a6-3274422b61a7"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "[(0.0, 1), (1.0, 0), (1.0, 0), (1.0, 0), (0.0, 1), (0.0, 1), (1.0, 1), (0.0, 1), (1.0, 1), (1.0, 1)]\n",
            "[(0.0, 1), (1.0, 1), (1.0, 1), (1.0, 1), (0.0, 1), (0.0, 1), (1.0, 1), (0.0, 1), (1.0, 1), (1.0, 1)]\n",
            "[(0.0, 1.0), (1.0, 0.0), (1.0, 0.0), (1.0, 0.0), (0.0, 1.0), (0.0, 0.0), (1.0, 0.0), (0.0, 1.0), (1.0, 0.0), (1.0, 1.0)]\n",
            "[(0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (1.0, 0.0), (0.0, 1.0), (0.0, 1.0), (1.0, 0.0), (0.0, 1.0), (1.0, 1.0), (1.0, 1.0)]\n",
            "lrAccuracy:0.627180527383\n",
            "svmAccuracy:0.514672075727\n",
            "nbAccuracy:0.580392156863\n",
            "dtAccuracy:0.648275862069\n"
          ]
        }
      ],
      "source": [
        "print lrLabelPrediction.take(10)\n",
        "print svmLabelPrediction.take(10)\n",
        "print nbLabelPrediction.take(10)\n",
        "print dtLabelPrediction.take(10)\n",
        " \n",
        "def accuracy(labelAndPrediction):\n",
        "    return labelAndPrediction.filter(lambda x: x[0]==x[1]).count()/float(labelAndPrediction.count())\n",
        "\n",
        "lrAccuracy = accuracy(lrLabelPrediction)\n",
        "svmAccuracy = accuracy(svmLabelPrediction)\n",
        "nbAccuracy = accuracy(nbLabelPrediction)\n",
        "dtAccuracy = accuracy(dtLabelPrediction)\n",
        "print 'lrAccuracy:{0}\\nsvmAccuracy:{1}\\nnbAccuracy:{2}\\ndtAccuracy:{3}'.format(lrAccuracy, svmAccuracy, nbAccuracy, dtAccuracy)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sdHYlpQQQUNg"
      },
      "source": [
        "### precision and recall"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5h17dLAUQUNg",
        "outputId": "68db77c8-110e-4084-ffd6-924524370b84"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "preciosn and recall for logistic regression (0.6220342964529011, 0.6975763962065332)\n",
            "preciosn and recall for svm (0.5140300935339569, 0.9989462592202318)\n",
            "preciosn and recall for naive bayes (0.6222222222222222, 0.46469968387776606)\n",
            "preciosn and recall for decision tree (0.6673200784094091, 0.6277660695468915)\n"
          ]
        }
      ],
      "source": [
        "def precisionAndRecall(labelAndPrediction):\n",
        "    TP = labelAndPrediction.filter(lambda x: x[0] == 1 and x[1] == 1).count()\n",
        "    FP = labelAndPrediction.filter(lambda x: x[0] == 0 and x[1] == 1).count()\n",
        "    FN = labelAndPrediction.filter(lambda x: x[0] == 1 and x[1] == 0).count()\n",
        "    return (float(TP)/(TP + FP), float(TP)/(TP + FN))\n",
        "\n",
        "print('preciosn and recall for logistic regression', precisionAndRecall(lrLabelPrediction))\n",
        "print('preciosn and recall for svm', precisionAndRecall(svmLabelPrediction))\n",
        "print('preciosn and recall for naive bayes', precisionAndRecall(nbLabelPrediction))\n",
        "print('preciosn and recall for decision tree', precisionAndRecall(dtLabelPrediction))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DbXYuAK9QUNg"
      },
      "source": [
        "## ROC and AUC"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "L9NAAbkQQUNg",
        "outputId": "0f02c4e1-e250-4f7c-87e6-17099f16a237"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "model\tArea under PR\tArea under ROC\n",
            "Logistic Regression (0.7374253598523676, 0.6252538830157423)\n",
            "SVM (0.7567586293858841, 0.5014181143280931)\n",
            "Naive Bayes (0.6808510815151734, 0.5835585110136261)\n",
            "Decision Tree (0.7430805993331199, 0.6488371887050935)\n"
          ]
        }
      ],
      "source": [
        "from pyspark.mllib.evaluation import BinaryClassificationMetrics\n",
        "\n",
        "# evaluate with built-in function\n",
        "def ROCAndAUC(labelAndPrediction):\n",
        "    metrics = BinaryClassificationMetrics(labelAndPrediction.map(lambda x:(float(x[1]), float(x[0]))))\n",
        "    return (metrics.areaUnderPR, metrics.areaUnderROC)\n",
        "\n",
        "print('model\\tArea under PR\\tArea under ROC')\n",
        "print('Logistic Regression',ROCAndAUC(lrLabelPrediction))\n",
        "print('SVM', ROCAndAUC(svmLabelPrediction))\n",
        "print('Naive Bayes', ROCAndAUC(nbLabelPrediction))\n",
        "print('Decision Tree', ROCAndAUC(dtLabelPrediction))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lI9R6wEGQUNh"
      },
      "source": [
        "# Improving model performance and tuning parameters"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "m1aLxgUxQUNh"
      },
      "source": [
        "## Feature standardization"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "uIdh_-LsQUNh",
        "outputId": "f5491fd2-36d2-4489-d807-db4b7173b96c"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "[  4.12258053e-01   2.76182319e+00   4.68230473e-01   2.14079926e-01\n",
            "   9.20623607e-02   4.92621604e-02   2.25510345e+00  -1.03750428e-01\n",
            "   0.00000000e+00   5.64227450e-02   2.12305612e-02   2.33778177e-01\n",
            "   2.75709037e-01   6.15551048e-01   6.60311021e-01   3.00770791e+01\n",
            "   3.97565923e-02   5.71659824e+03   1.78754564e+02   4.96064909e+00\n",
            "   1.72864050e-01   1.01220792e-01]\n",
            "[  1.09742442e-01   7.43008248e+01   4.12631699e-02   2.15334363e-02\n",
            "   9.21181745e-03   5.27493347e-03   3.25391871e+01   9.39698870e-02\n",
            "   0.00000000e+00   1.71774103e-03   2.07826348e-02   2.75483942e-03\n",
            "   3.68378892e+00   2.36679961e-01   2.24330712e-01   4.15878559e+02\n",
            "   3.81811688e-02   7.87733008e+07   3.22081162e+04   1.04530090e+01\n",
            "   3.35936340e-02   6.27753288e-03]\n"
          ]
        }
      ],
      "source": [
        "from pyspark.mllib.linalg.distributed import RowMatrix \n",
        "features = data.map(lambda d: d.features)\n",
        "matrix = RowMatrix(features)\n",
        "matrixSummary = matrix.computeColumnSummaryStatistics()\n",
        "print(matrixSummary.mean())\n",
        "print(matrixSummary.variance())\n",
        "# print(matrixSummary.min())\n",
        "# print(matrixSummary.max())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "scrolled": false,
        "id": "oMzU2Z3wQUNi",
        "outputId": "239c6c1c-f1f8-4247-a7aa-c093cb9b0da6"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "[ -3.28296418e-16   5.27355937e-16   1.74166237e-15   3.24393290e-16\n",
            "   1.70523318e-15  -9.47538489e-16  -9.67975700e-16  -1.45716772e-16\n",
            "   0.00000000e+00  -3.50414142e-16  -4.77048956e-18   3.43475248e-16\n",
            "  -3.67761377e-16   8.32667268e-17  -3.29597460e-17  -2.94729519e-15\n",
            "  -6.93889390e-18  -5.36029554e-16  -1.14491749e-15  -2.48065457e-16\n",
            "   2.00534034e-15  -1.72258041e-15]\n",
            "[ 1.  1.  1.  1.  1.  1.  1.  1.  0.  1.  1.  1.  1.  1.  1.  1.  1.  1.\n",
            "  1.  1.  1.  1.]\n"
          ]
        }
      ],
      "source": [
        "from pyspark.mllib.feature import StandardScaler\n",
        "scaler = StandardScaler(withMean = True, withStd = True).fit(features)\n",
        "\n",
        "# the following line is not allowed in distributed mode\n",
        "# scaledData = data.map(lambda d : LabeledPoint(d.label, scaler.transform(d.features)))\n",
        "labels = data.map(lambda d : d.label)\n",
        "scaledData = labels.zip(scaler.transform(features)).map(lambda x : LabeledPoint(x[0],x[1]))\n",
        "scaledFeatures = scaledData.map(lambda d: d.features)\n",
        "scaledMatrix = RowMatrix(scaledFeatures)\n",
        "scaledMatrixSummary = scaledMatrix.computeColumnSummaryStatistics()\n",
        "print(scaledMatrixSummary.mean())\n",
        "print(scaledMatrixSummary.variance())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "yElaUEmjQUNj",
        "outputId": "bc451b2e-9596-4325-baec-f071761715ef"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "('preciosn and recall', (0.62464046021093, 0.6865121180189674))\n",
            "('ROC and AUC', (0.7360360592298912, 0.6256956255557466))\n"
          ]
        }
      ],
      "source": [
        "# retrain the model with scaled data\n",
        "iterations = 10\n",
        "\n",
        "# logistic regression\n",
        "lr = LogisticRegressionWithLBFGS.train(scaledData, iterations)\n",
        "lrLP = scaledData.map(lambda x: (x.label, lr.predict(x.features)))\n",
        "print('preciosn and recall', precisionAndRecall(lrLP))\n",
        "print('Area under PR and ROC',ROCAndAUC(lrLP))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "UlCHM6PoQUNj"
      },
      "source": [
        "## Additional features"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "OprtP8wSQUNj",
        "outputId": "d391293e-a0a9-4f31-d3f0-de40b71ac59d"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "{u'\"recreation\"': 0, u'\"gaming\"': 7, u'\"arts_entertainment\"': 11, u'\"computer_internet\"': 1, u'\"?\"': 2, u'\"sports\"': 8, u'\"business\"': 3, u'\"health\"': 4, u'\"science_technology\"': 12, u'\"religion\"': 10, u'\"law_crime\"': 5, u'\"unknown\"': 9, u'\"culture_politics\"': 6, u'\"weather\"': 13}\n"
          ]
        }
      ],
      "source": [
        "categories = dataFile.map(lambda line:line.split('\\t')[3]).distinct().zipWithIndex().collect()\n",
        "categories = dict(categories)\n",
        "print(categories)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "VrxHUTZfQUNk",
        "outputId": "6f911dde-9e4c-49af-ec31-baa99f4ac17f"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "(0.0,[0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.789131,2.055555556,0.676470588,0.205882353,0.047058824,0.023529412,0.443783175,0.0,0.0,0.09077381,0.0,0.245831182,0.003883495,1.0,1.0,24.0,0.0,5424.0,170.0,8.0,0.152941176,0.079129575])\n"
          ]
        }
      ],
      "source": [
        "# broadCate = sc.broadcast(categories)\n",
        "def addCateFeature(line):\n",
        "    fields = line.strip().split('\\t')\n",
        "    cateFeature = [0] * len(categories)\n",
        "    idx = categories[fields[3]]\n",
        "    cateFeature[idx] = 1\n",
        "    label = int(fields[-1].replace('\"', ''))\n",
        "    for f in fields[4:-1]:\n",
        "        f = f.replace('\"', '')\n",
        "        if f != '?':\n",
        "            cateFeature.append(float(f))\n",
        "        else:\n",
        "            cateFeature.append(0.0)\n",
        "    return LabeledPoint(label, cateFeature)\n",
        "\n",
        "cateData = dataFile.map(addCateFeature)    \n",
        "print(cateData.first())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "scrolled": false,
        "id": "M70mPh0tQUNk",
        "outputId": "097b3f02-6d0f-4033-d93b-71313cf92685"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "(0.0,[-0.446421204794,-0.204182210579,-0.680752790425,2.72073665645,-0.270999069693,-0.0648775723926,-0.220526884579,-0.101894690972,-0.232727977095,-0.028494000387,-0.0991499193088,-0.381813223243,-0.201654052319,-0.0232621058984,1.1376473365,-0.0819355716929,1.02513981289,-0.0558635644254,-0.468893253129,-0.354305326308,-0.317535217236,0.33845079824,0.0,0.828822173315,-0.147268943346,0.229639823578,-0.141625969099,0.790238049918,0.717194729453,-0.297996816496,-0.20346257793,-0.0329672096969,-0.0487811297558,0.940069975117,-0.108698488525,-0.278820782314])\n"
          ]
        }
      ],
      "source": [
        "cateFeatures = cateData.map(lambda x : x.features)\n",
        "cateLabels = cateData.map(lambda x : x.label)\n",
        "cateScaler = StandardScaler(withMean = True, withStd = True).fit(cateFeatures)\n",
        "scaleCateData = cateLabels.zip(cateScaler.transform(cateFeatures)).map(lambda x : LabeledPoint(x[0], x[1]))\n",
        "print scaleCateData.first()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "scrolled": false,
        "id": "0vJKgAViQUNk",
        "outputId": "3df2e449-99e6-479d-f63c-9d09e8c88f39"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "('preciosn and recall', (0.67293997965412, 0.6970495258166491))\n",
            "('Area under PR and ROC', (0.7627499927624298, 0.6698640238141318))\n"
          ]
        }
      ],
      "source": [
        "iterations = 10\n",
        "\n",
        "# logistic regression\n",
        "lr = LogisticRegressionWithLBFGS.train(scaleCateData, iterations)\n",
        "lrLP = scaleCateData.map(lambda x: (x.label, lr.predict(x.features)))\n",
        "print('preciosn and recall', precisionAndRecall(lrLP))\n",
        "print('Area under PR and ROC',ROCAndAUC(lrLP))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "bkqGsYbTQUNl"
      },
      "source": [
        "## Using the correct form of data"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "scrolled": false,
        "id": "5w6lfZa5QUNl",
        "outputId": "a6c0d6b3-de79-4041-d055-4ddc69d35901"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "('preciosn and recall', (0.5916885212830341, 0.7726554267650158))\n",
            "('Area under PR and ROC', (0.7405222106704076, 0.6051384941549446))\n"
          ]
        }
      ],
      "source": [
        "# For naive bayesian MLlib implements a multinomial model\n",
        "# This model workson input in the form of non-zero count data. This can include a binary representation\n",
        "# of categorical features (such as the 1-of-k encoding covered previously) or frequency\n",
        "# data (such as the frequency of occurrences of words in a document\n",
        "def cateFeature(line):\n",
        "    fields = line.strip().split('\\t')\n",
        "    cateFeature = [0] * len(categories)\n",
        "    idx = categories[fields[3]]\n",
        "    cateFeature[idx] = 1\n",
        "    label = int(fields[-1].replace('\"', ''))\n",
        "    return LabeledPoint(label, cateFeature)\n",
        "\n",
        "nbData = dataFile.map(cateFeature)\n",
        "\n",
        "# naive bayes\n",
        "nbModel = NaiveBayes.train(nbData, 1.0) # 1.0 is the smoothing parameter\n",
        "nbLP = nbData.map(lambda x: (x.label, nbModel.predict(x.features)))\n",
        "print('preciosn and recall', precisionAndRecall(nbLP))\n",
        "print('Area under PR and ROC',ROCAndAUC(nbLP))"
      ]
    }
  ],
  "metadata": {
    "anaconda-cloud": {},
    "kernelspec": {
      "display_name": "Apache Toree - PySpark",
      "language": "python",
      "name": "apache_toree_pyspark"
    },
    "language_info": {
      "codemirror_mode": "text/x-ipython",
      "file_extension": ".py",
      "mimetype": "text/x-ipython",
      "name": "python",
      "pygments_lexer": "python",
      "version": "2.7.11\n"
    },
    "colab": {
      "provenance": []
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}