nb4-rdd-set/nb4-rdd-set.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Set operations on RDDs"
   ]
  },
  {
   "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 suporta muitas das operações que temos em conjuntos matemáticos, como união e interseção, mesmo quando os próprios RDDs não estão configurados corretamente. É importante observar que essas operações exigem que os RDDs que estão sendo operados sejam do mesmo tipo."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Definir operações é bastante simples de entender como funciona conforme o esperado. A única consideração vem do fato de que os RDDs não são conjuntos reais e, portanto, operações como a união de RDDs não removem duplicatas. Neste notebook, vamos dar uma breve olhada em `subtract`, `distinct` e `cartesian`."
   ]
  },
  {
   "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% do fornecido para a KDD Cup 1999, 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": 1,
   "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": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_file = \"./kddcup.data_10_percent.gz\"\n",
    "raw_data = sc.textFile(data_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Obtendo interações de ataque usando `subtract`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para fins ilustrativos, imagine que já temos nosso RDD com interações de não ataque (normal) de alguma análise anterior."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "normal_raw_data = raw_data.filter(lambda x: \"normal.\" in x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Podemos obter interações de ataque subtraindo as normais do RDD não filtrado original da seguinte maneira."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "attack_raw_data = raw_data.subtract(normal_raw_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Vamos fazer algumas contagens para verificar nossos resultados."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "All count in 3.694 secs\n"
     ]
    }
   ],
   "source": [
    "from time import time\n",
    "\n",
    "# count all\n",
    "t0 = time()\n",
    "raw_data_count = raw_data.count()\n",
    "tt = time() - t0\n",
    "print(\"All count in {} secs\".format(round(tt,3)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Normal count in 3.496 secs\n"
     ]
    }
   ],
   "source": [
    "# count normal\n",
    "t0 = time()\n",
    "normal_raw_data_count = normal_raw_data.count()\n",
    "tt = time() - t0\n",
    "print(\"Normal count in {} secs\".format(round(tt,3)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Attack count in 11.597 secs\n"
     ]
    }
   ],
   "source": [
    "# count attacks\n",
    "t0 = time()\n",
    "attack_raw_data_count = attack_raw_data.count()\n",
    "tt = time() - t0\n",
    "print(\"Attack count in {} secs\".format(round(tt,3)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 97278 normal interactions and 396743 attacks, from a total of 494021 interactions\n"
     ]
    }
   ],
   "source": [
    "print(\"There are {} normal interactions and {} attacks, \\\n",
    "from a total of {} interactions\".format(normal_raw_data_count,attack_raw_data_count,raw_data_count))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Então agora temos dois RDDs, um com interações normais e outro com ataques."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Combinações de protocolos e serviços usando `cartesian`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Podemos calcular o produto cartesiano entre dois RDDs usando a transformação cartesiana. Ele retorna todos os possíveis pares de elementos entre dois RDDs. Em nosso caso, vamos usá-lo para gerar todas as combinações possíveis entre serviço e protocolo em nossas interações de rede.\n",
    "\n",
    "Primeiro de tudo, precisamos isolar cada coleção de valores em dois RDDs separados. Para isso, usaremos `distinct` no conjunto de dados analisado por CSV. A partir da [descrição do conjunto de dados](http://kdd.ics.uci.edu/databases/kddcup99/kddcup.names) sabemos que o protocolo é a segunda coluna e o serviço é o terceiro (a tag é a última e não a primeira como aparece na página)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Então, primeiro, vamos pegar os protocolos."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['tcp', 'udp', 'icmp']"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csv_data = raw_data.map(lambda x: x.split(\",\"))\n",
    "protocols = csv_data.map(lambda x: x[1]).distinct()\n",
    "protocols.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Agora fazemos o mesmo pelos serviços."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['http',\n",
       " 'smtp',\n",
       " 'finger',\n",
       " 'domain_u',\n",
       " 'auth',\n",
       " 'telnet',\n",
       " 'ftp',\n",
       " 'eco_i',\n",
       " 'ntp_u',\n",
       " 'ecr_i',\n",
       " 'other',\n",
       " 'private',\n",
       " 'pop_3',\n",
       " 'ftp_data',\n",
       " 'rje',\n",
       " 'time',\n",
       " 'mtp',\n",
       " 'link',\n",
       " 'remote_job',\n",
       " 'gopher',\n",
       " 'ssh',\n",
       " 'name',\n",
       " 'whois',\n",
       " 'domain',\n",
       " 'login',\n",
       " 'imap4',\n",
       " 'daytime',\n",
       " 'ctf',\n",
       " 'nntp',\n",
       " 'shell',\n",
       " 'IRC',\n",
       " 'nnsp',\n",
       " 'http_443',\n",
       " 'exec',\n",
       " 'printer',\n",
       " 'efs',\n",
       " 'courier',\n",
       " 'uucp',\n",
       " 'klogin',\n",
       " 'kshell',\n",
       " 'echo',\n",
       " 'discard',\n",
       " 'systat',\n",
       " 'supdup',\n",
       " 'iso_tsap',\n",
       " 'hostnames',\n",
       " 'csnet_ns',\n",
       " 'pop_2',\n",
       " 'sunrpc',\n",
       " 'uucp_path',\n",
       " 'netbios_ns',\n",
       " 'netbios_ssn',\n",
       " 'netbios_dgm',\n",
       " 'sql_net',\n",
       " 'vmnet',\n",
       " 'bgp',\n",
       " 'Z39_50',\n",
       " 'ldap',\n",
       " 'netstat',\n",
       " 'urh_i',\n",
       " 'X11',\n",
       " 'urp_i',\n",
       " 'pm_dump',\n",
       " 'tftp_u',\n",
       " 'tim_i',\n",
       " 'red_i']"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "services = csv_data.map(lambda x: x[2]).distinct()\n",
    "services.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Uma lista mais longa neste caso."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Agora podemos fazer o produto cartesiano."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 198 combinations of protocol X service\n"
     ]
    }
   ],
   "source": [
    "product = protocols.cartesian(services).collect()\n",
    "print(\"There are {} combinations of protocol X service\".format(len(product)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Obviamente, para esses RDDs pequenos, não faz sentido usar o produto cartesiano Spark. Poderíamos ter coletado perfeitamente os valores depois de usar `distinct` e fazer o produto cartesiano localmente. Além disso, `distinct` e `cartesian` são operações dispendiosas, portanto devem ser usadas com cuidado quando os conjuntos de dados operacionais são grandes."
   ]
  }
 ],
 "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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}