nb6-rdd-key-value/nb6-rdd-key-value.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Trabalhando com RDDs de pares chave/valor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### [Baseado em \"Introduction to Spark with Python, by Jose A. Dianes\"](https://github.com/jadianes/spark-py-notebooks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "O Spark fornece funções específicas para lidar com RDDs cujos elementos são pares de chave/valor. Eles geralmente são usados para realizar agregações e outros processamentos por chave."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Neste caderno, mostraremos como, ao trabalhar com pares de chave/valor, podemos processar nosso conjunto de dados de interações de rede de uma maneira mais prática e poderosa do que a usada em cadernos anteriores. As agregações de pares chave/valor se mostrarão particularmente eficazes ao tentar explorar cada tipo de tag em nossos ataques de rede, de maneira individual."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Obtendo os dados e criando o RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Como fizemos em nosso primeiro notebook, usaremos o conjunto de dados reduzido (10 por cento) fornecido para a [KDD Cup 1999](http://kdd.ics.uci.edu/databases/kddcup99/kddcup99.html), contendo quase meio milhão de interações de rede. O arquivo é fornecido como um arquivo Gzip que será baixado localmente.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from urllib.request import urlretrieve\n",
    "\n",
    "f = urlretrieve(\"http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz\", \"kddcup.data_10_percent.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Criando um par RDD para tipos de interação"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Neste notebook, queremos fazer algumas análises exploratórias de dados em nosso conjunto de dados de interações de rede. Mais concretamente, queremos analisar cada tipo de interação de rede em termos de algumas de suas variáveis, como duração. Para fazer isso, primeiro precisamos criar o RDD adequado para isso, em que cada interação é analisada como uma linha CSV representando o valor e é colocada junto com sua tag correspondente como uma chave."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Normalmente, criamos RDDs de pares de chave/valor aplicando uma função usando `map` aos dados originais. Esta função retorna o par correspondente para um dado elemento RDD. Nós podemos proceder da seguinte forma."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "csv_data = raw_data.map(lambda x: x.split(\",\"))\n",
    "key_value_data = csv_data.map(lambda x: (x[41], x)) # x[41] contains the network interaction tag"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Agora temos nossos dados de par chave/valor prontos para serem usados. Vamos pegar o primeiro elemento para ver como é."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('normal.',\n",
       "  ['0',\n",
       "   'tcp',\n",
       "   'http',\n",
       "   'SF',\n",
       "   '181',\n",
       "   '5450',\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",
       "   '8',\n",
       "   '8',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '1.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '9',\n",
       "   '9',\n",
       "   '1.00',\n",
       "   '0.00',\n",
       "   '0.11',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   '0.00',\n",
       "   'normal.'])]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "key_value_data.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Agregações de dados com RDDs de par chave/valor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Podemos usar todas as transformações e ações disponíveis para RDDs normais com RDDs de par chave / valor. Nós só precisamos fazer as funções trabalharem com elementos de par. Além disso, o Spark fornece funções específicas para trabalhar com RDDs contendo elementos de par. Eles são muito semelhantes aos disponíveis para RDDs gerais. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Por exemplo, temos uma transformação `reduceByKey` que podemos usar da seguinte maneira para calcular a duração total de cada tipo de interação de rede."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('normal.', 21075991.0),\n",
       " ('buffer_overflow.', 2751.0),\n",
       " ('loadmodule.', 326.0),\n",
       " ('perl.', 124.0),\n",
       " ('neptune.', 0.0),\n",
       " ('smurf.', 0.0),\n",
       " ('guess_passwd.', 144.0),\n",
       " ('pod.', 0.0),\n",
       " ('teardrop.', 0.0),\n",
       " ('portsweep.', 1991911.0),\n",
       " ('ipsweep.', 43.0),\n",
       " ('land.', 0.0),\n",
       " ('ftp_write.', 259.0),\n",
       " ('back.', 284.0),\n",
       " ('imap.', 72.0),\n",
       " ('satan.', 64.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": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "key_value_duration = csv_data.map(lambda x: (x[41], float(x[0]))) \n",
    "durations_by_key = key_value_duration.reduceByKey(lambda x, y: x + y)\n",
    "\n",
    "durations_by_key.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Temos uma ação de contagem específica para pares chave/valor. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {'normal.': 97278,\n",
       "             'buffer_overflow.': 30,\n",
       "             'loadmodule.': 9,\n",
       "             'perl.': 3,\n",
       "             'neptune.': 107201,\n",
       "             'smurf.': 280790,\n",
       "             'guess_passwd.': 53,\n",
       "             'pod.': 264,\n",
       "             'teardrop.': 979,\n",
       "             'portsweep.': 1040,\n",
       "             'ipsweep.': 1247,\n",
       "             'land.': 21,\n",
       "             'ftp_write.': 8,\n",
       "             'back.': 2203,\n",
       "             'imap.': 12,\n",
       "             'satan.': 1589,\n",
       "             'phf.': 4,\n",
       "             'nmap.': 231,\n",
       "             'multihop.': 7,\n",
       "             'warezmaster.': 20,\n",
       "             'warezclient.': 1020,\n",
       "             'spy.': 2,\n",
       "             'rootkit.': 10})"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "counts_by_key = key_value_data.countByKey()\n",
    "counts_by_key"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Utilizando o `combineByKey`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Essa é a mais geral das funções de agregação por chave. A maioria dos outros combinadores por chave é implementada usando-o. Podemos pensar nisso como o equivalente a `aggregate`, pois permite que o usuário retorne valores que não são do mesmo tipo que nossos dados de entrada."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Por exemplo, podemos usá-lo para calcular durações médias por tipo, como segue."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'normal.': (21075991.0, 97278),\n",
       " 'buffer_overflow.': (2751.0, 30),\n",
       " 'loadmodule.': (326.0, 9),\n",
       " 'perl.': (124.0, 3),\n",
       " 'neptune.': (0.0, 107201),\n",
       " 'smurf.': (0.0, 280790),\n",
       " 'guess_passwd.': (144.0, 53),\n",
       " 'pod.': (0.0, 264),\n",
       " 'teardrop.': (0.0, 979),\n",
       " 'portsweep.': (1991911.0, 1040),\n",
       " 'ipsweep.': (43.0, 1247),\n",
       " 'land.': (0.0, 21),\n",
       " 'ftp_write.': (259.0, 8),\n",
       " 'back.': (284.0, 2203),\n",
       " 'imap.': (72.0, 12),\n",
       " 'satan.': (64.0, 1589),\n",
       " 'phf.': (18.0, 4),\n",
       " 'nmap.': (0.0, 231),\n",
       " 'multihop.': (1288.0, 7),\n",
       " 'warezmaster.': (301.0, 20),\n",
       " 'warezclient.': (627563.0, 1020),\n",
       " 'spy.': (636.0, 2),\n",
       " 'rootkit.': (1008.0, 10)}"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sum_counts = key_value_duration.combineByKey(\n",
    "    (lambda x: (x, 1)), # the initial value, with value x and count 1\n",
    "    (lambda acc, value: (acc[0]+value, acc[1]+1)), # how to combine a pair value with the accumulator: sum value, and increment count\n",
    "    (lambda acc1, acc2: (acc1[0]+acc2[0], acc1[1]+acc2[1])) # combine accumulators\n",
    ")\n",
    "\n",
    "sum_counts.collectAsMap()"
   ]
  }
 ],
 "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.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}