intermediate_notebooks/examples/blazingsql/vs_pyspark_netflow.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "d0hJ4z8rBOFC"
   },
   "source": [
    "# BlazingSQL vs. Apache Spark \n",
    "\n",
    "Below we have one of our popular workloads running with [BlazingSQL](https://blazingsql.com/) + [RAPIDS AI](https://rapids.ai) and then running the entire ETL phase again, only this time with Apache Spark + PySpark.\n",
    "\n",
    "In this notebook, we will cover: \n",
    "- How to read and query csv files with cuDF and BlazingSQL.\n",
    "- How BlazingSQL compares against Apache Spark (analyzing over 20M records)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### BlazingSQL install check\n",
    "The next cell checks that you have BlazingSQL installed, and offers to install it if not (making sure the notebook will run as expected)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"You've got BlazingSQL set up perfectly! Let's get started with SQL in RAPIDS AI!\""
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import sys \n",
    "# point import path notebooks-contrib/utils\n",
    "sys.path.append('../../../utils/')\n",
    "from sql_check import bsql_start\n",
    "# check that BlazingSQL is installed\n",
    "bsql_start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Download Data\n",
    "This cell will check if you have the data for this demo, and, if you don't, will download it for you."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2020-01-21 17:27:12--  https://blazingsql-colab.s3.amazonaws.com/netflow_data/nf-chunk2.csv\n",
      "Resolving blazingsql-colab.s3.amazonaws.com (blazingsql-colab.s3.amazonaws.com)... 52.216.115.3\n",
      "Connecting to blazingsql-colab.s3.amazonaws.com (blazingsql-colab.s3.amazonaws.com)|52.216.115.3|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 2725056295 (2.5G) [text/csv]\n",
      "Saving to: ‘../../../data/blazingsql/nf-chunk2.csv’\n",
      "\n",
      "nf-chunk2.csv       100%[===================>]   2.54G  43.5MB/s    in 56s     \n",
      "\n",
      "2020-01-21 17:28:14 (46.0 MB/s) - ‘../../../data/blazingsql/nf-chunk2.csv’ saved [2725056295/2725056295]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "\n",
    "# relative path to data folder\n",
    "data_dir = '../../../data/blazingsql/'\n",
    "# file name\n",
    "fn = 'nf-chunk2.csv'\n",
    "\n",
    "# does folder exist?\n",
    "if not os.path.exists(data_dir):\n",
    "    print('creating blazingsql directory')\n",
    "    # create folder\n",
    "    os.system('mkdir ../../data/blazingsql')\n",
    "\n",
    "# do we have music file?\n",
    "if not os.path.isfile(data_dir + fn):\n",
    "    # save nf-chunk2 to data folder, may take a few minutes to download (21,526,138 records)\n",
    "    !wget -P ../../../data/blazingsql https://blazingsql-colab.s3.amazonaws.com/netflow_data/nf-chunk2.csv\n",
    "else:\n",
    "    print(\"You've got the data!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create BlazingContext\n",
    "You can think of the BlazingContext much like a Spark Context, this is where information such as FileSystems you have registered and Tables you have created will be stored. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 35
    },
    "colab_type": "code",
    "id": "ojm_V-WAtz0f",
    "outputId": "a46625f4-1494-4a13-eb13-2f38efd80ccf"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "BlazingContext ready\n"
     ]
    }
   ],
   "source": [
    "import cudf\n",
    "from blazingsql import BlazingContext\n",
    "# start up BlazingSQL\n",
    "bc = BlazingContext()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "OTEaAsp2_zmf"
   },
   "source": [
    "## BlazingSQL + cuDF \n",
    "Data in hand, we can test the preformance of cuDF and BlazingSQL on this dataset. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 52
    },
    "colab_type": "code",
    "id": "rirBsYQU3NH5",
    "outputId": "51ced2b1-b930-4173-bbfa-09672e751d3f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 3.75 s, sys: 1.19 s, total: 4.94 s\n",
      "Wall time: 4.93 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# Load CSVs into GPU DataFrames (GDF)\n",
    "netflow_gdf = cudf.read_csv(data_dir + fn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 52
    },
    "colab_type": "code",
    "id": "zCzLEFfB3N4k",
    "outputId": "10ff9097-2736-423e-969d-de75983fbdda"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 4.11 ms, sys: 23 µs, total: 4.13 ms\n",
      "Wall time: 3.32 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<pyblazing.apiv2.context.BlazingTable at 0x7f9230411278>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "# Create BlazingSQL table from GDF - There is no copy in this process\n",
    "bc.create_table('netflow', netflow_gdf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 295
    },
    "colab_type": "code",
    "id": "umBG2Tp0wbQx",
    "outputId": "0975395e-7f5b-4244-afa3-45c8658ce61c"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.98 s, sys: 514 ms, total: 2.49 s\n",
      "Wall time: 1.95 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# define the query\n",
    "query = '''\n",
    "        select\n",
    "            a.firstSeenSrcIp as source,\n",
    "            a.firstSeenDestIp as destination,\n",
    "            count(a.firstSeenDestPort) as targetPorts,\n",
    "            sum(a.firstSeenSrcTotalBytes) as bytesOut,\n",
    "            sum(a.firstSeenDestTotalBytes) as bytesIn,\n",
    "            sum(a.durationSeconds) as durationSeconds,\n",
    "            min(parsedDate) as firstFlowDate,\n",
    "            max(parsedDate) as lastFlowDate,\n",
    "            count(*) as attemptCount\n",
    "        from \n",
    "            netflow a\n",
    "        group by\n",
    "            a.firstSeenSrcIp,\n",
    "            a.firstSeenDestIp\n",
    "            '''\n",
    "\n",
    "# query the table (returns cudf dataframe)\n",
    "result_gdf = bc.sql(query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "48_W2v8q_zmq",
    "outputId": "db0394f1-e082-49b0-c477-e3bba8d3d0f4"
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>source</th>\n",
       "      <th>destination</th>\n",
       "      <th>targetPorts</th>\n",
       "      <th>bytesOut</th>\n",
       "      <th>bytesIn</th>\n",
       "      <th>durationSeconds</th>\n",
       "      <th>firstFlowDate</th>\n",
       "      <th>lastFlowDate</th>\n",
       "      <th>attemptCount</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>172.10.1.162</td>\n",
       "      <td>10.0.0.11</td>\n",
       "      <td>87</td>\n",
       "      <td>39628</td>\n",
       "      <td>53983</td>\n",
       "      <td>24</td>\n",
       "      <td>2013-04-03 06:50:13</td>\n",
       "      <td>2013-04-03 14:58:35</td>\n",
       "      <td>87</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>172.30.2.60</td>\n",
       "      <td>10.0.0.9</td>\n",
       "      <td>82</td>\n",
       "      <td>34839</td>\n",
       "      <td>47716</td>\n",
       "      <td>134</td>\n",
       "      <td>2013-04-03 06:48:47</td>\n",
       "      <td>2013-04-03 12:12:37</td>\n",
       "      <td>82</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>172.30.1.56</td>\n",
       "      <td>172.0.0.1</td>\n",
       "      <td>25</td>\n",
       "      <td>3330</td>\n",
       "      <td>3240</td>\n",
       "      <td>67</td>\n",
       "      <td>2013-04-03 01:59:09</td>\n",
       "      <td>2013-04-03 22:05:39</td>\n",
       "      <td>25</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>172.10.1.234</td>\n",
       "      <td>10.0.0.5</td>\n",
       "      <td>104</td>\n",
       "      <td>47287</td>\n",
       "      <td>64750</td>\n",
       "      <td>18</td>\n",
       "      <td>2013-04-03 06:53:55</td>\n",
       "      <td>2013-04-03 15:11:07</td>\n",
       "      <td>104</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>10.1.0.76</td>\n",
       "      <td>172.10.1.82</td>\n",
       "      <td>1</td>\n",
       "      <td>633</td>\n",
       "      <td>392</td>\n",
       "      <td>0</td>\n",
       "      <td>2013-04-03 09:55:05</td>\n",
       "      <td>2013-04-03 09:55:05</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>172.30.1.85</td>\n",
       "      <td>10.0.0.8</td>\n",
       "      <td>84</td>\n",
       "      <td>37828</td>\n",
       "      <td>52864</td>\n",
       "      <td>3</td>\n",
       "      <td>2013-04-03 06:48:21</td>\n",
       "      <td>2013-04-03 12:06:53</td>\n",
       "      <td>84</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>172.30.1.10</td>\n",
       "      <td>10.0.0.12</td>\n",
       "      <td>69</td>\n",
       "      <td>31042</td>\n",
       "      <td>43044</td>\n",
       "      <td>25</td>\n",
       "      <td>2013-04-03 06:48:01</td>\n",
       "      <td>2013-04-03 12:11:40</td>\n",
       "      <td>69</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>172.30.1.201</td>\n",
       "      <td>172.0.0.1</td>\n",
       "      <td>29</td>\n",
       "      <td>2610</td>\n",
       "      <td>2610</td>\n",
       "      <td>0</td>\n",
       "      <td>2013-04-03 00:26:46</td>\n",
       "      <td>2013-04-03 23:06:00</td>\n",
       "      <td>29</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>172.30.2.125</td>\n",
       "      <td>10.0.0.9</td>\n",
       "      <td>69</td>\n",
       "      <td>30701</td>\n",
       "      <td>41558</td>\n",
       "      <td>341</td>\n",
       "      <td>2013-04-03 06:50:50</td>\n",
       "      <td>2013-04-03 12:12:37</td>\n",
       "      <td>69</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>172.10.1.89</td>\n",
       "      <td>10.0.0.5</td>\n",
       "      <td>112</td>\n",
       "      <td>51222</td>\n",
       "      <td>70260</td>\n",
       "      <td>24</td>\n",
       "      <td>2013-04-03 06:48:24</td>\n",
       "      <td>2013-04-03 15:17:39</td>\n",
       "      <td>112</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "         source  destination  targetPorts  bytesOut  bytesIn  durationSeconds  \\\n",
       "0  172.10.1.162    10.0.0.11           87     39628    53983               24   \n",
       "1   172.30.2.60     10.0.0.9           82     34839    47716              134   \n",
       "2   172.30.1.56    172.0.0.1           25      3330     3240               67   \n",
       "3  172.10.1.234     10.0.0.5          104     47287    64750               18   \n",
       "4     10.1.0.76  172.10.1.82            1       633      392                0   \n",
       "5   172.30.1.85     10.0.0.8           84     37828    52864                3   \n",
       "6   172.30.1.10    10.0.0.12           69     31042    43044               25   \n",
       "7  172.30.1.201    172.0.0.1           29      2610     2610                0   \n",
       "8  172.30.2.125     10.0.0.9           69     30701    41558              341   \n",
       "9   172.10.1.89     10.0.0.5          112     51222    70260               24   \n",
       "\n",
       "         firstFlowDate         lastFlowDate  attemptCount  \n",
       "0  2013-04-03 06:50:13  2013-04-03 14:58:35            87  \n",
       "1  2013-04-03 06:48:47  2013-04-03 12:12:37            82  \n",
       "2  2013-04-03 01:59:09  2013-04-03 22:05:39            25  \n",
       "3  2013-04-03 06:53:55  2013-04-03 15:11:07           104  \n",
       "4  2013-04-03 09:55:05  2013-04-03 09:55:05             1  \n",
       "5  2013-04-03 06:48:21  2013-04-03 12:06:53            84  \n",
       "6  2013-04-03 06:48:01  2013-04-03 12:11:40            69  \n",
       "7  2013-04-03 00:26:46  2013-04-03 23:06:00            29  \n",
       "8  2013-04-03 06:50:50  2013-04-03 12:12:37            69  \n",
       "9  2013-04-03 06:48:24  2013-04-03 15:17:39           112  "
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# how's it look?\n",
    "result_gdf.head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "6PXbjW1hTxrD"
   },
   "source": [
    "## Apache Spark\n",
    "The cell below installs Apache Spark ([PySpark](https://spark.apache.org/docs/latest/api/python/index.html))."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "pnEEvVEtT8xi"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting pyspark\n",
      "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/87/21/f05c186f4ddb01d15d0ddc36ef4b7e3cedbeb6412274a41f26b55a650ee5/pyspark-2.4.4.tar.gz (215.7MB)\n",
      "\u001b[K     |################################| 215.7MB 77.0MB/s eta 0:00:011\n",
      "\u001b[?25hCollecting py4j==0.10.7 (from pyspark)\n",
      "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/e3/53/c737818eb9a7dc32a7cd4f1396e787bd94200c3997c72c1dbe028587bd76/py4j-0.10.7-py2.py3-none-any.whl (197kB)\n",
      "\u001b[K     |################################| 204kB 72.4MB/s eta 0:00:01\n",
      "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
      "  Building wheel for pyspark (setup.py) ... \u001b[?25ldone\n",
      "\u001b[?25h  Created wheel for pyspark: filename=pyspark-2.4.4-py2.py3-none-any.whl size=216130387 sha256=7879a54a037a812709763c4abf7d3d85b5b9b9f8ac6278785942767dc8032f54\n",
      "  Stored in directory: /root/.cache/pip/wheels/ab/09/4d/0d184230058e654eb1b04467dbc1292f00eaa186544604b471\n",
      "Successfully built pyspark\n",
      "Installing collected packages: py4j, pyspark\n",
      "Successfully installed py4j-0.10.7 pyspark-2.4.4\n"
     ]
    }
   ],
   "source": [
    "# installs Spark (2.4.4 Jan 2020)\n",
    "!pip install pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "W3-XmZkz_zmw"
   },
   "source": [
    "#### PyBlazing vs PySpark\n",
    "With everything installed we can launch a SparkSession and see how BlazingSQL stacks up."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 51
    },
    "colab_type": "code",
    "id": "nioEt2MqT9B0",
    "outputId": "f75b9823-5dbd-45b1-9282-562d3d6ddaf0"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 73.4 ms, sys: 27.2 ms, total: 101 ms\n",
      "Wall time: 6.59 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "#I copied this cell's snippet from another Google Colab by Luca Canali here: https://colab.research.google.com/github/LucaCanali/sparkMeasure/blob/master/examples/SparkMeasure_Jupyter_Colab_Example.ipynb\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Create Spark Session\n",
    "# This example uses a local cluster, you can modify master to use  YARN or K8S if available \n",
    "# This example downloads sparkMeasure 0.13 for scala 2_11 from maven central\n",
    "\n",
    "spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .master(\"local[*]\") \\\n",
    "        .appName(\"PySpark Netflow Benchmark code\") \\\n",
    "        .config(\"spark.jars.packages\",\"ch.cern.sparkmeasure:spark-measure_2.11:0.13\")  \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "G8XSppQiUdLY"
   },
   "source": [
    "### Load & Query Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 51
    },
    "colab_type": "code",
    "id": "ZSLuSYSOUDtf",
    "outputId": "2b93169b-63c5-4c46-da14-af87645bf51b"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 20.4 ms, sys: 33.4 ms, total: 53.8 ms\n",
      "Wall time: 48.8 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# load CSV into Spark\n",
    "netflow_df = spark.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load(data_dir+fn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 51
    },
    "colab_type": "code",
    "id": "iT3BwLn8UDwE",
    "outputId": "4eeff800-489f-4230-adb9-f3a1c16ede66"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.87 ms, sys: 0 ns, total: 1.87 ms\n",
      "Wall time: 28.6 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# create table for querying\n",
    "netflow_df.createOrReplaceTempView('netflow')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 493
    },
    "colab_type": "code",
    "id": "9SBhahA5UD2k",
    "outputId": "accc1938-6470-44df-ab7f-70058c755b2b"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+---------------+-----------+--------+-------+---------------+-------------------+-------------------+------------+\n",
      "|      source|    destination|targetPorts|bytesOut|bytesIn|durationSeconds|      firstFlowDate|       lastFlowDate|attemptCount|\n",
      "+------------+---------------+-----------+--------+-------+---------------+-------------------+-------------------+------------+\n",
      "| 172.10.1.13|239.255.255.250|         15|    2975|      0|              6|2013-04-03 06:36:19|2013-04-03 06:36:27|          15|\n",
      "|172.30.1.204|239.255.255.250|          8|    1750|      0|              6|2013-04-03 06:36:13|2013-04-03 06:36:20|           8|\n",
      "| 172.30.2.86|      172.0.0.1|          1|     540|      0|              2|2013-04-03 06:36:09|2013-04-03 06:36:09|           1|\n",
      "|172.30.1.246|      172.0.0.1|         29|    2610|   2610|              0|2013-04-03 00:26:46|2013-04-03 23:06:00|          29|\n",
      "| 172.30.1.51|239.255.255.250|         16|    3850|      0|             18|2013-04-03 06:35:22|2013-04-03 06:44:08|          16|\n",
      "+------------+---------------+-----------+--------+-------+---------------+-------------------+-------------------+------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 12.1 ms, sys: 11.2 ms, total: 23.4 ms\n",
      "Wall time: 21.1 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# define the same query run tested on blazingsql above\n",
    "query = '''\n",
    "        SELECT\n",
    "            a.firstSeenSrcIp as source,\n",
    "            a.firstSeenDestIp as destination,\n",
    "            count(a.firstSeenDestPort) as targetPorts,\n",
    "            SUM(a.firstSeenSrcTotalBytes) as bytesOut,\n",
    "            SUM(a.firstSeenDestTotalBytes) as bytesIn,\n",
    "            SUM(a.durationSeconds) as durationSeconds,\n",
    "            MIN(parsedDate) as firstFlowDate,\n",
    "            MAX(parsedDate) as lastFlowDate,\n",
    "            COUNT(*) as attemptCount\n",
    "        FROM\n",
    "            netflow a\n",
    "        GROUP BY\n",
    "            a.firstSeenSrcIp,\n",
    "            a.firstSeenDestIp\n",
    "        '''\n",
    "\n",
    "# query with Spark\n",
    "edges_df = spark.sql(query)\n",
    "\n",
    "# set/display results\n",
    "edges_df.show(5)"
   ]
  }
 ],
 "metadata": {
  "accelerator": "GPU",
  "colab": {
   "collapsed_sections": [],
   "name": "vs_pyspark_netflow.ipynb",
   "provenance": [],
   "toc_visible": true
  },
  "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}