WatsonStudio_QueryCOS.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val accessKey = \"\"\n",
    "val secretKey = \"\"\n",
    "val bucketName = \"streamingdata\"\n",
    "val endpoint = \"s3-api.us-geo.objectstorage.service.networklayer.com\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Waiting for a Spark session to start..."
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Waiting for a Spark session to start..."
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "s3Url = s3a://streamingdata/\n",
       "schema = StructType(StructField(InvoiceNo,LongType,true), StructField(StockCode,LongType,true), StructField(Description,StringType,true), StructField(Quantity,ShortType,true), StructField(InvoiceDate,LongType,true), StructField(UnitPrice,DoubleType,true), StructField(CustomerID,IntegerType,true), StructField(Country,StringType,true), StructField(LineNo,ShortType,true), StructField(InvoiceTime,StringType,true), StructField(StoreID,ShortType,true), StructField(TransactionID,StringType,true))\n",
       "df = [InvoiceNo: bigint, StockCode: bigint ... 10 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[InvoiceNo: bigint, StockCode: bigint ... 10 more fields]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.hadoopConfiguration.set(\"fs.s3.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\");\n",
    "sc.hadoopConfiguration.set(\"fs.s3a.access.key\", accessKey)\n",
    "sc.hadoopConfiguration.set(\"fs.s3a.secret.key\", secretKey)\n",
    "sc.hadoopConfiguration.set(\"fs.s3a.endpoint\", endpoint)\n",
    "\n",
    "val s3Url = s\"s3a://${bucketName}/\"\n",
    "\n",
    "import org.apache.spark.sql.types._\n",
    "\n",
    "val schema = (new StructType()\n",
    "    .add(\"InvoiceNo\", LongType)\n",
    "    .add(\"StockCode\", LongType)\n",
    "    .add(\"Description\", StringType)\n",
    "    .add(\"Quantity\", ShortType)\n",
    "    .add(\"InvoiceDate\", LongType)\n",
    "    .add(\"UnitPrice\", DoubleType)\n",
    "    .add(\"CustomerID\", IntegerType)\n",
    "    .add(\"Country\", StringType)\n",
    "    .add(\"LineNo\", ShortType)\n",
    "    .add(\"InvoiceTime\", StringType)\n",
    "    .add(\"StoreID\", ShortType)\n",
    "    .add(\"TransactionID\", StringType))\n",
    "\n",
    "var df = spark.read.schema(schema).json(s\"${s3Url}/data/*\")\n",
    "\n",
    "println(df.count())\n",
    "df.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><td>5373891</td><td>22197</td><td>SMALL POPCORN HOLDER</td><td>10</td><td>1535636460000</td><td>0.85</td><td>17757</td><td>United Kingdom</td><td>15</td><td>13:41:00</td><td>0</td><td>5373891150180830</td></tr>\n",
       "<tr><td>5373891</td><td>20752</td><td>BLUE POLKADOT WASHING UP GLOVES</td><td>4</td><td>1535636460000</td><td>2.1</td><td>17757</td><td>United Kingdom</td><td>17</td><td>13:41:00</td><td>0</td><td>5373891170180830</td></tr>\n",
       "<tr><td>5405261</td><td>22624</td><td>IVORY KITCHEN SCALES</td><td>1</td><td>1535636460000</td><td>8.5</td><td>14606</td><td>United Kingdom</td><td>3</td><td>13:41:00</td><td>0</td><td>540526130180830</td></tr>\n",
       "<tr><td>5405261</td><td>22199</td><td>FRYING PAN RED RETROSPOT</td><td>1</td><td>1535636460000</td><td>4.25</td><td>14606</td><td>United Kingdom</td><td>8</td><td>13:41:00</td><td>0</td><td>540526180180830</td></tr>\n",
       "<tr><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td><td>NULL</td></tr>\n",
       "</table>"
      ],
      "text/plain": [
       "+---------+-------+---------------------------------+------+---------------+------+-------+----------------+------+----------+------+------------------+\n",
       "| 5373891 | 22197 | SMALL POPCORN HOLDER            | 10   | 1535636460000 | 0.85 | 17757 | United Kingdom | 15   | 13:41:00 | 0    | 5373891150180830 |\n",
       "| 5373891 | 20752 | BLUE POLKADOT WASHING UP GLOVES | 4    | 1535636460000 | 2.1  | 17757 | United Kingdom | 17   | 13:41:00 | 0    | 5373891170180830 |\n",
       "| 5405261 | 22624 | IVORY KITCHEN SCALES            | 1    | 1535636460000 | 8.5  | 14606 | United Kingdom | 3    | 13:41:00 | 0    | 540526130180830  |\n",
       "| 5405261 | 22199 | FRYING PAN RED RETROSPOT        | 1    | 1535636460000 | 4.25 | 14606 | United Kingdom | 8    | 13:41:00 | 0    | 540526180180830  |\n",
       "| NULL    | NULL  | NULL                            | NULL | NULL          | NULL | NULL  | NULL           | NULL | NULL     | NULL | NULL             |\n",
       "+---------+-------+---------------------------------+------+---------------+------+-------+----------------+------+----------+------+------------------+"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import org.apache.spark.sql.functions._"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "df = [InvoiceNo: bigint, StockCode: bigint ... 11 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[InvoiceNo: bigint, StockCode: bigint ... 11 more fields]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df =  ( df\n",
    "            .filter(col(\"InvoiceNo\").isNotNull)\n",
    "            .filter(col(\"CustomerID\").isNotNull)\n",
    "            // Create a column 'Cancelled' which has the values 1=Cancelled, 0=Not Cancelled\n",
    "            .withColumn(\"Cancelled\",\n",
    "              when(col(\"InvoiceNO\").startsWith(\"C\"), lit(1)).otherwise(lit(0))\n",
    "            )\n",
    "            .withColumn(\"UnitPrice\", abs($\"UnitPrice\"))\n",
    "            .withColumn(\"Quantity\", abs($\"Quantity\"))\n",
    " )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><td>5373891</td><td>22197</td><td>SMALL POPCORN HOLDER</td><td>10</td><td>1535636460000</td><td>0.85</td><td>17757</td><td>United Kingdom</td><td>15</td><td>13:41:00</td><td>0</td><td>5373891150180830</td><td>0</td></tr>\n",
       "<tr><td>5373891</td><td>20752</td><td>BLUE POLKADOT WASHING UP GLOVES</td><td>4</td><td>1535636460000</td><td>2.1</td><td>17757</td><td>United Kingdom</td><td>17</td><td>13:41:00</td><td>0</td><td>5373891170180830</td><td>0</td></tr>\n",
       "<tr><td>5405261</td><td>22624</td><td>IVORY KITCHEN SCALES</td><td>1</td><td>1535636460000</td><td>8.5</td><td>14606</td><td>United Kingdom</td><td>3</td><td>13:41:00</td><td>0</td><td>540526130180830</td><td>0</td></tr>\n",
       "<tr><td>5405261</td><td>22199</td><td>FRYING PAN RED RETROSPOT</td><td>1</td><td>1535636460000</td><td>4.25</td><td>14606</td><td>United Kingdom</td><td>8</td><td>13:41:00</td><td>0</td><td>540526180180830</td><td>0</td></tr>\n",
       "<tr><td>5412141</td><td>22431</td><td>WATERING CAN BLUE ELEPHANT</td><td>6</td><td>1535636460000</td><td>1.95</td><td>15570</td><td>United Kingdom</td><td>16</td><td>13:41:00</td><td>0</td><td>5412141160180830</td><td>0</td></tr>\n",
       "</table>"
      ],
      "text/plain": [
       "+---------+-------+---------------------------------+-----+---------------+------+-------+----------------+-----+----------+-----+------------------+-----+\n",
       "| 5373891 | 22197 | SMALL POPCORN HOLDER            | 10  | 1535636460000 | 0.85 | 17757 | United Kingdom | 15  | 13:41:00 | 0   | 5373891150180830 | 0   |\n",
       "| 5373891 | 20752 | BLUE POLKADOT WASHING UP GLOVES | 4   | 1535636460000 | 2.1  | 17757 | United Kingdom | 17  | 13:41:00 | 0   | 5373891170180830 | 0   |\n",
       "| 5405261 | 22624 | IVORY KITCHEN SCALES            | 1   | 1535636460000 | 8.5  | 14606 | United Kingdom | 3   | 13:41:00 | 0   | 540526130180830  | 0   |\n",
       "| 5405261 | 22199 | FRYING PAN RED RETROSPOT        | 1   | 1535636460000 | 4.25 | 14606 | United Kingdom | 8   | 13:41:00 | 0   | 540526180180830  | 0   |\n",
       "| 5412141 | 22431 | WATERING CAN BLUE ELEPHANT      | 6   | 1535636460000 | 1.95 | 15570 | United Kingdom | 16  | 13:41:00 | 0   | 5412141160180830 | 0   |\n",
       "+---------+-------+---------------------------------+-----+---------------+------+-------+----------------+-----+----------+-----+------------------+-----+"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "countryIndexer = strIdx_5b722d61fa70\n",
       "assembler = vecAssembler_65c8cfdde2f5\n",
       "trainingData = [InvoiceNo: bigint, StockCode: bigint ... 11 more fields]\n",
       "testData = [InvoiceNo: bigint, StockCode: bigint ... 11 more fields]\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "[InvoiceNo: bigint, StockCode: bigint ... 11 more fields]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "text/plain": [
       "dtc: org.apache.spark.ml.class...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.classification.DecisionTreeClassificationModel\n",
    "import org.apache.spark.ml.classification.DecisionTreeClassifier\n",
    "import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator\n",
    "import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer, VectorAssembler}\n",
    "\n",
    "val countryIndexer = new StringIndexer()\n",
    "  .setInputCol(\"Country\")\n",
    "  .setOutputCol(\"indexedCountry\")\n",
    "  .fit(df)\n",
    "\n",
    "val assembler = new VectorAssembler()\n",
    "  .setInputCols(Array(\"indexedCountry\", \"StockCode\", \"Quantity\", \"UnitPrice\"))\n",
    "  .setOutputCol(\"features\")\n",
    "\n",
    "val Array(trainingData, testData) = df.randomSplit(Array(0.7, 0.3))\n",
    "\n",
    "val dtc = new DecisionTreeClassifier()\n",
    "  .setLabelCol(\"Cancelled\")\n",
    "  .setFeaturesCol(\"features\")\n",
    "\n",
    "val pipeline = new Pipeline()\n",
    "  .setStages(Array(countryIndexer, assembler, dtc))\n",
    "\n",
    "val model = pipeline.fit(trainingData)\n",
    "\n",
    "// Make predictions.\n",
    "val predictions = model.transform(testData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------+---------+---------+----------+\n",
      "|   TransactionID|       Country|StockCode|Cancelled|Prediction|\n",
      "+----------------+--------------+---------+---------+----------+\n",
      "| 536537160180830|United Kingdom|    22183|        0|       0.0|\n",
      "| 536537120180830|United Kingdom|    22333|        0|       0.0|\n",
      "| 536538110180830|United Kingdom|    21466|        0|       0.0|\n",
      "| 536538120180830|United Kingdom|    21467|        0|       0.0|\n",
      "|5365381130180830|United Kingdom|    21690|        0|       0.0|\n",
      "+----------------+--------------+---------+---------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "// Select example rows to display.\n",
    "predictions.select(\"TransactionID\", \"Country\", \"StockCode\", \"Cancelled\", \"Prediction\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test Error = 0.0\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "evaluator = mcEval_7fa7a079f23e\n",
       "accuracy = 1.0\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "1.0"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val evaluator = new MulticlassClassificationEvaluator()\n",
    "  .setLabelCol(\"Cancelled\")\n",
    "  .setPredictionCol(\"prediction\")\n",
    "  .setMetricName(\"accuracy\")\n",
    "\n",
    "val accuracy = evaluator.evaluate(predictions)\n",
    "println(\"Test Error = \" + (1.0 - accuracy))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Scala 2.11 with Spark 2.3 (YARN Client Mode)",
   "language": "scala",
   "name": "scala-spark23"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "2.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}