pipeline_engineering_pyspark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark \n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "findspark.init() \n",
    "    \n",
    "from pyspark.sql import *\n",
    "# from pyspark import SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext('local', 'HandsOn PySpark')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "visitors = [10,3,35,25,41,9,29]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_visitors = sc.parallelize(visitors)\n",
    "# df_visitors_yearly = df_visitors.map(lambda x:  x * 365).collect()\n",
    "# print(df_visitors_yearly)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_visitors_yearly = df_visitors.map(lambda x:  x * 365).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[3650, 1095, 12775, 9125, 14965, 3285, 10585]\n"
     ]
    }
   ],
   "source": [
    "type(df_visitors)\n",
    "print(df_visitors_yearly)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !pip install --upgrade pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# dir(df_visitors)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !cp /Users/director/Downloads/spark_rd.md ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark SQL My sample\") \\\n",
    "    .master(\"local[*]\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "text_file = spark.read.text(\"spark_rd.md\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "32"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "text_file.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(value='# Apache Spark')"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "text_file.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines_with_spark = text_file.filter(text_file.value.contains(\"Spark\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|               value|\n",
      "+--------------------+\n",
      "|      # Apache Spark|\n",
      "|Spark is a unifie...|\n",
      "|rich set of highe...|\n",
      "+--------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "lines_with_spark.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "text_file.filter(text_file.value.contains(\"Spark\")).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_src = \"http://archive.ics.uci.edu/ml/machine-learning-databases/kddcup99-mld/kddcup.data.gz\"\n",
    "import urllib.request"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# f = urllib.request.urlretrieve(data_src, \"kddcup.data.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.1.64:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.4</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>HandsOn PySpark</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local appName=HandsOn PySpark>"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_cup_data = sc.textFile(\"./kddcup.data.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "contains_normal = raw_cup_data.filter(lambda line: \"normal.\" in line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "972781"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "contains_normal.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "split_file = raw_cup_data.map(lambda line: line.split(\",\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "# split_file.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "sampled = raw_cup_data.sample(False, 0.1, 42) # , 10%, seed: 42"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "contains_norma_sample = sampled.map(lambda x: x.split(\",\")).filter(lambda x: \"normal\"in x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "18.889207124710083\n"
     ]
    }
   ],
   "source": [
    "t0 = time()\n",
    "num_sampled = contains_norma_sample.count()\n",
    "duration = time() - t0\n",
    "print(duration)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "contains_norma = raw_cup_data.map(lambda x: x.split(\",\")).filter(lambda x: \"normal\"in x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "36.53693389892578\n"
     ]
    }
   ],
   "source": [
    "t1 = time()\n",
    "full_count = contains_norma.count()\n",
    "duration = time() - t1\n",
    "print(duration)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_in_memory = raw_cup_data.takeSample(False, 20, 42)\n",
    "contains_normal_py = [line.split(\",\") for line in data_in_memory if \"normal\" in line]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(contains_normal_py)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Lines without Normal\n",
    "normal_sample = sampled.filter(lambda line: \"normal\" in line)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "non_normal_sample = sampled.subtract(normal_sample)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "490705"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sampled.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "97404"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "normal_sample.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "393301"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "non_normal_sample.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "# AGGREGATIONS\n",
    "csv = raw_cup_data.map(lambda x: x.split(\",\"))\n",
    "normal_data = csv.filter(lambda x: x[41] == \"normal.\")\n",
    "# normal_data = csv.filter(lambda x: \"normal\" in x)\n",
    "duration = normal_data.map(lambda x: int(x[0]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 0, 0]"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "duration.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "211895753"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "total_duration = duration.reduce(lambda x,y: x+y)\n",
    "total_duration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "217.82472416710442"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "avg_value = total_duration/(normal_data.count())\n",
    "avg_value"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### FASTER AVERAGE COMPUTATIONS WITH AGGREGATE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "217.82472416710442"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "duration_count = duration.aggregate(\n",
    "    (0,0),\n",
    "    (lambda db, new_value: (db[0] + new_value, db[1] + 1)),\n",
    "    (lambda db1, db2: (db1[0] + db2[0],db1[1] + db2[1]))\n",
    ")\n",
    "\n",
    "duration_count[0]/duration_count[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "first_4 = [1,2,3,4]\n",
    "seqOp = (lambda x,y: (x[0] + y, x[1] + 1))\n",
    "combOp = (lambda x,y: (x[0] + y[0], x[1] + y[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(10, 4)"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "res_4 = sc.parallelize(first_4).aggregate((0,0), seqOp, combOp)\n",
    "res_4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(30, 5)"
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fives = [2,4,6,8,10]\n",
    "res_5 = sc.parallelize(fives).aggregate((0,0), seqOp, combOp) # (0,0) - init value, seqOp - functin=on\n",
    "res_5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### PIVOT TABLING WITH KEY-VALUE PAIRED DATA POINTS\n",
    "Grouping data by key-values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('normal.',\n",
       "  ['0',\n",
       "   'tcp',\n",
       "   'http',\n",
       "   'SF',\n",
       "   '215',\n",
       "   '45076',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '1',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0',\n",
       "   '1',\n",
       "   '1',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '1.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0',\n",
       "   '0',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   'normal.'])]"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kv = csv.map(lambda x: (x[41], x))\n",
    "kv.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('normal.', 211895753.0),\n",
       " ('buffer_overflow.', 2751.0),\n",
       " ('loadmodule.', 326.0),\n",
       " ('perl.', 124.0),\n",
       " ('neptune.', 2.0),\n",
       " ('smurf.', 0.0),\n",
       " ('guess_passwd.', 144.0),\n",
       " ('pod.', 0.0),\n",
       " ('teardrop.', 0.0),\n",
       " ('portsweep.', 24257982.0),\n",
       " ('ipsweep.', 13049.0),\n",
       " ('land.', 0.0),\n",
       " ('ftp_write.', 259.0),\n",
       " ('back.', 284.0),\n",
       " ('imap.', 72.0),\n",
       " ('satan.', 500.0),\n",
       " ('phf.', 18.0),\n",
       " ('nmap.', 0.0),\n",
       " ('multihop.', 1288.0),\n",
       " ('warezmaster.', 301.0),\n",
       " ('warezclient.', 627563.0),\n",
       " ('spy.', 636.0),\n",
       " ('rootkit.', 1008.0)]"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kv_duration = csv.map(lambda x: (x[41], float(x[0]))).reduceByKey(lambda x,y: x + y)\n",
    "kv_duration.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {'normal.': 972781,\n",
       "             'buffer_overflow.': 30,\n",
       "             'loadmodule.': 9,\n",
       "             'perl.': 3,\n",
       "             'neptune.': 1072017,\n",
       "             'smurf.': 2807886,\n",
       "             'guess_passwd.': 53,\n",
       "             'pod.': 264,\n",
       "             'teardrop.': 979,\n",
       "             'portsweep.': 10413,\n",
       "             'ipsweep.': 12481,\n",
       "             'land.': 21,\n",
       "             'ftp_write.': 8,\n",
       "             'back.': 2203,\n",
       "             'imap.': 12,\n",
       "             'satan.': 15892,\n",
       "             'phf.': 4,\n",
       "             'nmap.': 2316,\n",
       "             'multihop.': 7,\n",
       "             'warezmaster.': 20,\n",
       "             'warezclient.': 1020,\n",
       "             'spy.': 2,\n",
       "             'rootkit.': 10})"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kv.countByKey()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data Analysis with MLlib"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Computing Summary Statistics\n",
    "from pyspark.mllib.linalg import Vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "# rdd = sc.parallelize([Vectors.dense()])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.stat import Statistics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [],
   "source": [
    "duration = csv.map(lambda x: [int(x[0])])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "48.34243046395863"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "summary = Statistics.colStats(duration)\n",
    "summary.mean()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "723.3298112546713"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from math import sqrt\n",
    "sqrt(summary.variance()[0]) #std dev"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([58329.])"
      ]
     },
     "execution_count": 69,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "summary.max()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([0.])"
      ]
     },
     "execution_count": 70,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "summary.min()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using Pearson and Spearman to discover correlation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[ 1.        ,  0.00890383,  0.30144701],\n",
       "       [ 0.00890383,  1.        , -0.19510495],\n",
       "       [ 0.30144701, -0.19510495,  1.        ]])"
      ]
     },
     "execution_count": 72,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "metrics = csv.map(lambda x: [x[0], x[4], x[5]])\n",
    "spearman_corr = Statistics.corr(metrics, method=\"spearman\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spearman_corr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[1.00000000e+00, 4.12205545e-02, 2.03915936e-02],\n",
       "       [4.12205545e-02, 1.00000000e+00, 2.39337570e-04],\n",
       "       [2.03915936e-02, 2.39337570e-04, 1.00000000e+00]])"
      ]
     },
     "execution_count": 73,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pearson_corr = Statistics.corr(metrics, method=\"pearson\")\n",
    "pearson_corr"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Testing your Hypotheses on large datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import Vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Chi squared test summary:\n",
      "method: pearson\n",
      "degrees of freedom = 4 \n",
      "statistic = 0.5852136752136753 \n",
      "pValue = 0.9646925263439344 \n",
      "No presumption against null hypothesis: observed follows the same distribution as expected..\n"
     ]
    }
   ],
   "source": [
    "visitors_freq = Vectors.dense(0.13,0.61,0.8,0.5,0.3)\n",
    "print(Statistics.chiSqTest(visitors_freq))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### SPARK SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Row, SQLContext\n",
    "rows = csv.map(lambda p: Row(duration=int(p[0]), protocol=p[1], service=p[2]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql_context = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = sql_context.createDataFrame(rows)\n",
    "df.registerTempTable(\"rdd\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|duration|\n",
      "+--------+\n",
      "|   10217|\n",
      "|   11610|\n",
      "|   13724|\n",
      "|   10934|\n",
      "|   12026|\n",
      "+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql_context.sql(\"\"\"SELECT duration FROM rdd WHERE protocol = 'tcp' AND duration > 2000\"\"\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using Spark DSL to build queries for structured data operations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|duration|\n",
      "+--------+\n",
      "|   10217|\n",
      "|   11610|\n",
      "|   13724|\n",
      "|   10934|\n",
      "|   12026|\n",
      "+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"duration\").filter(df.duration > 2000).filter(df.protocol==\"tcp\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}