Jupyter/M04-DataManipulation/M04G-SparkSQL.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FK3_qvIrh3Lz"
      },
      "source": [
        "![Cloud-First](../image/CloudFirst.png)\n",
        "\n",
        "\n",
        "# SIT742: Modern Data Science\n",
        "**(Module: Big Data Manipulation)**\n",
        "\n",
        "---\n",
        "- Materials in this module include resources collected from various open-source online repositories.\n",
        "- You are free to use, change and distribute this package.\n",
        "- If you found any issue/bug for this document, please submit an issue at [tulip-lab/sit742](https://github.com/tulip-lab/sit742/issues)\n",
        "\n",
        "\n",
        "Prepared by **SIT742 Teaching Team**\n",
        "\n",
        "---\n",
        "\n",
        "\n",
        "## Session 4G: SparkSQL and Data Understanding\n",
        "---\n",
        "\n",
        "### Table of Content\n",
        "\n",
        "Part A: SparkSQL\n",
        "\n",
        "1. Loading in a DataFrame\n",
        "\n",
        "2. Creating SQLContext\n",
        "\n",
        "3. Creating DataFrame\n",
        "\n",
        "4. Grouping Aggregation\n",
        "\n",
        "5. Running SQL Queries\n",
        "\n",
        "Part B: SparkSQL Application\n",
        "\n",
        "6.  Getting the data and creating the RDD\n",
        "\n",
        "7. Getting a Data Frame\n",
        "\n",
        "8. Queries as DataFrame operations\n",
        "\n",
        "\n",
        "---\n",
        "\n",
        "## Introduction ##\n",
        "\n",
        "This notebook will introduce Spark capabilities to deal with data in a structured way. Basically, everything turns around the concept of *Data Frame* and using *SQL language* to query them. We will see how the data frame abstraction, very popular in other data analytics ecosystems (e.g. R and Python/Pandas), it is very powerful when performing exploratory data analysis.\n",
        "\n",
        "In fact, it is very easy to express data queries when used together with the SQL language. Moreover, Spark distributes this column-based data structure transparently, in order to make the querying process as efficient as possible.      \n",
        "\n",
        "This lab session will assume that you have uploaded two data files into cloud, and note down the address:\n",
        "- mtcars.csv\n",
        "- kddcup_data_10_percent-d8e1d.gz\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "LtLolzbVY3j9"
      },
      "source": [
        "# Part A: SparkSQL"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CwTeJHXPh3L2"
      },
      "source": [
        "### 1.Loading in a DataFrame\n",
        "To create a Spark DataFrame we load an external DataFrame, called `mtcars`. This DataFrame includes 32 observations on 11 variables.\n",
        "\n",
        "[, 1]\tmpg\tMiles/(US) --> gallon  \n",
        "[, 2]\tcyl\t--> Number of cylinders  \n",
        "[, 3]\tdisp\t--> Displacement (cu.in.)  \n",
        "[, 4]\thp -->\tGross horsepower  \n",
        "[, 5]\tdrat -->\tRear axle ratio  \n",
        "[, 6]\twt -->\tWeight (lb/1000)  \n",
        "[, 7]\tqsec -->\t1/4 mile time  \n",
        "[, 8]\tvs -->\tV/S  \n",
        "[, 9]\tam -->\tTransmission (0 = automatic, 1 = manual)  \n",
        "[,10]\tgear -->\tNumber of forward gears  \n",
        "[,11]\tcarb -->\tNumber of carburetors  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "2IVBZtPYh3L3"
      },
      "outputs": [],
      "source": [
        "!pip install wget"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "K4pcGZbRh3L7"
      },
      "outputs": [],
      "source": [
        "import wget\n",
        "\n",
        "link_to_data = 'https://raw.githubusercontent.com/tulip-lab/sit742/develop/Jupyter/data/mtcars.csv'\n",
        "DataSet = wget.download(link_to_data)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ZgQBUBGmh3L-"
      },
      "outputs": [],
      "source": [
        "import pandas as pd\n",
        "mtcars = pd.read_csv('mtcars.csv')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8bn_g86Bh3MA",
        "scrolled": true
      },
      "outputs": [],
      "source": [
        "mtcars.head()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VLqidgEsh3MC"
      },
      "source": [
        "### 2.Initialize SQLContext\n",
        "To work with dataframes we need a SQLContext which is created using `SQLContext(sc)`. SQLContext uses SparkContext which was the main entry point for Spark functionality, named `sc`.\n",
        "\n",
        "A SparkContext represents the connection to a Spark cluster, and can be used to create RDDs, accumulators and broadcast variables on that cluster. Only one SparkContext may be active per JVM. You must stop() the active SparkContext before creating a new one."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "hiTfYwOpinyN"
      },
      "outputs": [],
      "source": [
        "#update local version of the package catalog\n",
        "!apt-get update\n",
        "# install Java8\n",
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
        "# download spark 3.3.3\n",
        "!wget -q https://archive.apache.org/dist/spark/spark-3.3.3/spark-3.3.3-bin-hadoop3.tgz\n",
        "# unzip it\n",
        "!tar xf spark-3.3.3-bin-hadoop3.tgz\n",
        "# install findspark\n",
        "!pip install -q findspark\n",
        "\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark-3.3.3-bin-hadoop3\"\n",
        "\n",
        "import findspark\n",
        "findspark.init()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "JpFoyri0h3MD"
      },
      "outputs": [],
      "source": [
        "from pyspark import SparkContext\n",
        "from pyspark.sql import SQLContext\n",
        "\n",
        "#The getOrCreate function for the SparkContext may be used to get or instantiate a SparkContext and register it as a singleton object.\n",
        "sc = SparkContext.getOrCreate()\n",
        "\n",
        "#SQLContext is the entry point for working with structured data (rows and columns) in Spark, in Spark 1.x.\n",
        "#As of Spark 2.0, this is replaced by SparkSession.\n",
        "#However, we are keeping the class here for backward compatibility.\n",
        "#A SQLContext can be used create DataFrame, register DataFrame as tables, execute SQL over tables, cache tables, and read parquet files\n",
        "sqlContext = SQLContext(sc)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "doJ6p8Shh3MF"
      },
      "source": [
        "### 3.Creating Spark DataFrames\n",
        "With SQLContext and a loaded local DataFrame, we create a Spark DataFrame:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "9ECTDKYTh3MG"
      },
      "outputs": [],
      "source": [
        "#Creates a DataFrame from an RDD, a list or a pandas.DataFrame\n",
        "sdf = sqlContext.createDataFrame(mtcars.to_dict('records'))\n",
        "#Prints out the schema in the tree format.\n",
        "sdf.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jE7OEjWLh3MH"
      },
      "source": [
        "You can also directly load this csv file into a Spark DataFrame."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "GN2d6lXBh3MJ"
      },
      "outputs": [],
      "source": [
        "#Sample 1\n",
        "#Returns a DataFrameReader that can be used to read data in as a DataFrame.\n",
        "sdf2 = sqlContext.read.csv(\"mtcars.csv\")\n",
        "print(sdf2)\n",
        "\n",
        "\n",
        "#Sample 2 to define a specific format\n",
        "#Specifies the input data source format.\n",
        "sdf3=sqlContext.read.format('csv').load('mtcars.csv')\n",
        "print(sdf3)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FB64EqmZh3MK"
      },
      "source": [
        "#### (3a) Displays the content of the DataFrame\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "-vnHoYeWh3ML",
        "scrolled": true
      },
      "outputs": [],
      "source": [
        "sdf.show(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "l5EyxC_8h3MM"
      },
      "source": [
        "#### (3b) Selecting columns"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "22cyMzU2h3MN",
        "scrolled": true
      },
      "outputs": [],
      "source": [
        "sdf.select('mpg').show(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kDAw-uUth3MO"
      },
      "source": [
        "#### (3c)  Filtering Data\n",
        "Filter the DataFrame to only retain rows with `mpg` less than 18"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "uwTmPCmnh3MP"
      },
      "outputs": [],
      "source": [
        "sdf.filter(sdf['mpg'] < 18).show(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "D5oJ2FWjh3MR"
      },
      "source": [
        "#### (3d)  Operating on Columns\n",
        "Spark also provides a number of functions that can directly applied to columns for data processing and aggregation. The example below shows the use of basic arithmetic functions to convert lb to metric ton."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "iEsLIAAvh3MS"
      },
      "outputs": [],
      "source": [
        "# A new clomun name 'wtTon' is created and its value equals 'wt' * 0.45\n",
        "sdf.withColumn('wtTon', sdf['wt'] * 0.45).show(6)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "U7jQbGjah3MU"
      },
      "outputs": [],
      "source": [
        "sdf.show(6)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sTYersd9h3MW"
      },
      "source": [
        "###4.Grouping, Aggregation\n",
        "Spark DataFrames support a number of commonly used functions to aggregate data after grouping. For example we can compute the average weight of cars by their cylinders as shown below:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "jV5KWdv2h3MX"
      },
      "outputs": [],
      "source": [
        "sdf.groupby(['cyl'])\\\n",
        ".agg({\"wt\": \"AVG\"})\\\n",
        ".show(5)\n",
        "\n",
        "#It also equals the below line:\n",
        "sdf.groupby(['cyl']).agg({\"wt\": \"AVG\"}).show(5)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "xRCXT1ykh3Md"
      },
      "outputs": [],
      "source": [
        "# We can also sort the output from the aggregation to get the most common cars\n",
        "car_counts = sdf.groupby(['cyl'])\\\n",
        ".agg({\"wt\": \"count\"})\\\n",
        ".sort(\"count(wt)\", ascending=False)\\\n",
        ".show(5)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "z8T2LxMih3Mg"
      },
      "source": [
        "### 5.Running SQL Queries from Spark DataFrames\n",
        "A Spark DataFrame can also be registered as a temporary table in Spark SQL and registering a DataFrame as a table allows you to run SQL queries over its data. The `sql` function enables applications to run SQL queries programmatically and returns the result as a DataFrame.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "nS2N_w0ph3Mh"
      },
      "outputs": [],
      "source": [
        "# Register this DataFrame as a table.\n",
        "sdf.registerTempTable(\"cars\")\n",
        "\n",
        "# SQL statements can be run by using the sql method\n",
        "highgearcars = sqlContext.sql(\"SELECT gear FROM cars WHERE cyl >= 4 AND cyl <= 9\")\n",
        "highgearcars.show(6)\n",
        ""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BUsyGXlYh3Mj"
      },
      "source": [
        "## Part B: SparkSQL Application"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3S65v5gsh3Mk"
      },
      "source": [
        "### 6.Getting the data and creating the RDD"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yMfA3XGDh3Ml"
      },
      "source": [
        "As we did in previous notebooks, we will use the reduced dataset (10 percent) provided for the [KDD Cup 1999](http://kdd.ics.uci.edu/databases/kddcup99/kddcup99.html), containing nearly half million network interactions. The file is provided as a Gzip file that we will download from the GitHub.\n",
        "\n",
        "You can also use the urllib to download the orginal Gzip file as below codes:\n",
        "```\n",
        "import urllib\n",
        "f = urllib.urlretrieve (\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")\n",
        "```"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "OT5mRphWh3Mm"
      },
      "outputs": [],
      "source": [
        "import wget\n",
        "link_to_data = 'https://github.com/tulip-lab/sit742/raw/develop/Jupyter/data/kddcup.gz'\n",
        "DataSet = wget.download(link_to_data, out='kdd.gz')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "7TtPuauZh3Mn"
      },
      "outputs": [],
      "source": [
        "!ls -l\n",
        "!pwd"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "rGBn15krh3Mp"
      },
      "outputs": [],
      "source": [
        "data_file = \"kdd.gz\"\n",
        "\n",
        "#textFile is used to read a text file from HDFS,\n",
        "#a local file system (available on all nodes),\n",
        "#or any Hadoop-supported file system URI, and return it as an RDD of Strings.\n",
        "#In here, we assume that the file \"kdd.gz\" haved been in the HDFS\n",
        "raw_data = sc.textFile(data_file).cache()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wBHMNiQfh3Mq"
      },
      "source": [
        "### 7.Getting a Data Frame"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "M0L5eo9qh3Mr"
      },
      "source": [
        "A Spark `DataFrame` is a distributed collection of data organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R or Pandas. They can be constructed from a wide array of sources such as a existing RDD in our case."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "h7p2cOPXh3Mr"
      },
      "source": [
        "The entry point into all SQL functionality in Spark is the `SQLContext` class. To create a basic instance, all we need is a `SparkContext` reference. Since we are running Spark in shell mode (using pySpark) we can use the global context object `sc` for this purpose.    "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "e-n3zqZdh3Ms"
      },
      "outputs": [],
      "source": [
        "from pyspark.sql import SQLContext\n",
        "sqlContext = SQLContext(sc)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "38t_qtE5h3Mu"
      },
      "source": [
        "#### (7a) Inferring the schema"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "HOTn61x2h3Mv"
      },
      "source": [
        "With a `SQLContext`, we are ready to create a `DataFrame` from our existing RDD. But first we need to tell Spark SQL the schema in our data.   "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3VLh3U1Xh3Mv"
      },
      "source": [
        "Spark SQL can convert an RDD of `Row` objects to a `DataFrame`. Rows are constructed by passing a list of key/value pairs as *kwargs* to the `Row` class. The keys define the column names, and the types are inferred by looking at the first row. Therefore, it is important that there is no missing data in the first row of the RDD in order to properly infer the schema."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "us4USMoRh3Mw"
      },
      "source": [
        "In our case, we first need to split the comma separated data, and then use the information in KDD's 1999 task description to obtain the [column names](http://kdd.ics.uci.edu/databases/kddcup99/kddcup.names).  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "HQmuTXp_h3Mw"
      },
      "outputs": [],
      "source": [
        "from pyspark.sql import Row\n",
        "\n",
        "\n",
        "#Map is used to return an iterator that applies function to every item of iterable, yielding the results.\n",
        "#If additional iterable arguments are passed, function must take that many arguments and is applied to the items from all iterables in parallel.\n",
        "#With multiple iterables, the iterator stops when the shortest iterable is exhausted.\n",
        "csv_data = raw_data.map(lambda l: l.split(\",\"))\n",
        "row_data = csv_data.map(lambda p: Row(\n",
        "    duration=int(p[0]),\n",
        "    protocol_type=p[1],\n",
        "    service=p[2],\n",
        "    flag=p[3],\n",
        "    src_bytes=int(p[4]),\n",
        "    dst_bytes=int(p[5])\n",
        "    )\n",
        ")"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "fh3QDIN7h3My",
        "collapsed": true
      },
      "outputs": [],
      "source": [
        "help(sqlContext)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ReJOY-xkh3Mz"
      },
      "source": [
        "Once we have our RDD of `Row` we can infer and register the schema.  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lyVjzbF4h3M0"
      },
      "outputs": [],
      "source": [
        "#Creates a DataFrame from an RDD, a list or a pandas.DataFrame.\n",
        "interactions_df = sqlContext.createDataFrame(row_data)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "I-CjscEDh3M1"
      },
      "outputs": [],
      "source": [
        "#Registers this RDD as a temporary table using the given name.\n",
        "interactions_df.registerTempTable(\"interactions\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XxoWQb44h3M2"
      },
      "source": [
        "Now we can run SQL queries over our data frame that has been registered as a table.  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Rq2JNE3th3M3"
      },
      "outputs": [],
      "source": [
        "# Select tcp network interactions with more than 1 second duration and no transfer from destination\n",
        "# The sqlContext is uesd to returns a DataFrame representing the result of the given query.\n",
        "tcp_interactions = sqlContext.sql(\"\"\"\n",
        "    SELECT duration, dst_bytes FROM interactions WHERE protocol_type = 'tcp' AND duration > 1000 AND dst_bytes = 0\n",
        "\"\"\")\n",
        "tcp_interactions.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "35qDetV_h3M4"
      },
      "source": [
        "The results of SQL queries are RDDs and support all the normal RDD operations.  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "YCwynY9Nh3M4"
      },
      "outputs": [],
      "source": [
        "# Output duration together with dst_bytes\n",
        "# Return a new RDD by applying a function to each element of this RDD.\n",
        "tcp_interactions_out = tcp_interactions.rdd.map(lambda p: \"Duration: {}, Dest. bytes: {}\".format(p.duration, p.dst_bytes))\n",
        "for ti_out in tcp_interactions_out.collect():\n",
        "    print(ti_out)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "z1j30aDVh3M6"
      },
      "source": [
        "We can easily have a look at our data frame schema using `printSchema`.  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5G2gKHWsh3M6"
      },
      "outputs": [],
      "source": [
        "interactions_df.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PwyuGRLPh3M8"
      },
      "source": [
        "### 8.Queries as `DataFrame` operations"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JRtyQgqdh3M-"
      },
      "source": [
        "Spark `DataFrame` provides a domain-specific language for structured data manipulation. This language includes methods we can concatenate in order to do selection, filtering, grouping, etc. For example, let's say we want to count how many interactions are there for each protocol type. We can proceed as follows.  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "PirwiZv4h3M_"
      },
      "outputs": [],
      "source": [
        "from time import time\n",
        "\n",
        "t0 = time()\n",
        "interactions_df.select(\"protocol_type\", \"duration\", \"dst_bytes\").groupBy(\"protocol_type\").count().show()\n",
        "tt = time() - t0\n",
        "\n",
        "print(\"Query performed in {} seconds\".format(round(tt,3)))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Put6CVcBh3NA"
      },
      "source": [
        "Now imagine that we want to count how many interactions last more than 1 second, with no data transfer from destination, grouped by protocol type. We can just add to filter calls to the previous.   "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Mj8HXB-6h3NB"
      },
      "outputs": [],
      "source": [
        "t0 = time()\n",
        "interactions_df.select(\"protocol_type\", \"duration\", \"dst_bytes\").filter(interactions_df.duration>1000).filter(interactions_df.dst_bytes==0).groupBy(\"protocol_type\").count().show()\n",
        "tt = time() - t0\n",
        "\n",
        "print(\"Query performed in {} seconds\".format(round(tt,3)))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YFdljE5fh3NC"
      },
      "source": [
        "We can use this to perform some [exploratory data analysis](http://en.wikipedia.org/wiki/Exploratory_data_analysis). Let's count how many attack and normal interactions we have. First we need to add the label column to our data.    "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "bCQx7bX3h3ND"
      },
      "outputs": [],
      "source": [
        "def get_label_type(label):\n",
        "    if label!=\"normal.\":\n",
        "        return \"attack\"\n",
        "    else:\n",
        "        return \"normal\"\n",
        "\n",
        "row_labeled_data = csv_data.map(lambda p: Row(\n",
        "    duration=int(p[0]),\n",
        "    protocol_type=p[1],\n",
        "    service=p[2],\n",
        "    flag=p[3],\n",
        "    src_bytes=int(p[4]),\n",
        "    dst_bytes=int(p[5]),\n",
        "    label=get_label_type(p[41])\n",
        "    )\n",
        ")\n",
        "interactions_labeled_df = sqlContext.createDataFrame(row_labeled_data)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VaIxW6znh3NF"
      },
      "source": [
        "This time we don't need to register the schema since we are going to use the OO query interface.  "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "o_z8_n1Kh3NF"
      },
      "source": [
        "Let's check the previous actually works by counting attack and normal data in our data frame.  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Q31tE2sgh3NF"
      },
      "outputs": [],
      "source": [
        "t0 = time()\n",
        "interactions_labeled_df.select(\"label\").groupBy(\"label\").count().show()\n",
        "tt = time() - t0\n",
        "\n",
        "print(\"Query performed in {} seconds\".format(round(tt,3)))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-POnaJAUh3NH"
      },
      "source": [
        "Now we want to count them by label and protocol type, in order to see how important the protocol type is to detect when an interaction is or not an attack.  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "LfQgEhhvh3NI"
      },
      "outputs": [],
      "source": [
        "t0 = time()\n",
        "interactions_labeled_df.select(\"label\", \"protocol_type\").groupBy(\"label\", \"protocol_type\").count().show()\n",
        "tt = time() - t0\n",
        "\n",
        "print(\"Query performed in {} seconds\".format(round(tt,3)))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XmryXB3kh3NK"
      },
      "source": [
        "At first sight it seems that *udp* interactions are in lower proportion between network attacks versus other protocol types.  "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "p8Ub4yuEh3NL"
      },
      "source": [
        "And we can do much more sofisticated groupings. For example, add to the previous a \"split\" based on data transfer from target.   "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8u0t0an8h3NL"
      },
      "outputs": [],
      "source": [
        "t0 = time()\n",
        "interactions_labeled_df.select(\"label\", \"protocol_type\", \"dst_bytes\").groupBy(\"label\", \"protocol_type\", interactions_labeled_df.dst_bytes==0).count().show()\n",
        "tt = time() - t0\n",
        "\n",
        "print(\"Query performed in {} seconds\".format(round(tt,3)))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IIq3qh9zh3NM"
      },
      "source": [
        "We see how relevant is this new split to determine if a network interaction is an attack.  "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "NCbeFsx5h3NM"
      },
      "source": [
        "We will stop here, but we can see how powerfull this type of queries are in order to explore our data. Actually we can replicate all the splits we saw in previous notebooks, when introducing classification trees, just by selecting, groping, and filtering our dataframe. For a more detailed (but less real-world) list of Spark's `DataFrame` operations and data sources, have a look at the oficial documentation [here](https://spark.apache.org/docs/latest/sql-programming-guide.html#dataframe-operations).    "
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "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.7.3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}