00- Analise-Spark-Uber.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# <font color='blue'>Big Data Real-Time Analytics com Python e Spark</font>\n",
    "\n",
    "# <font color='blue'>Capítulo 7</font>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### *********** Atenção: *********** \n",
    "Utilize Java JDK 11 e Apache Spark 2.4.2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Caso receba mensagem de erro \"name 'sc' is not defined\", interrompa o pyspark e apague o diretório metastore_db no mesmo diretório onde está este Jupyter notebook*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Mini-Projeto 1 - Analisando Dados do Uber com Spark "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dataset: https://github.com/fivethirtyeight/uber-tlc-foil-response"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Esse conjunto de dados contém dados de mais de 4,5 milhões de captações Uber na cidade de Nova York de abril a setembro de 2014 e 14,3 milhões de captações Uber de janeiro a junho de 2015.\n",
    "\n",
    "Dados em nível de viagem sobre 10 outras empresas de veículos de aluguel (FHV) bem como dados agregados para 329 empresas de FHV, também estão incluídos.\n",
    "\n",
    "Todos os arquivos foram recebidos em 3 de agosto, 15 de setembro e 22 de setembro de 2015."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1- Quantos são e quais são as bases de carros do Uber (onde os carros ficam esperando passageiros)?\n",
    "\n",
    "2- Qual o total de veículos que passaram pela base B02617?\n",
    "\n",
    "3- Qual o total de corridas por base? Apresente de forma decrescente."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pandas import read_csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando um objeto Pandas\n",
    "uberFile = read_csv(\"data/uber.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pandas.core.frame.DataFrame"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(uberFile)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "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>dispatching_base_number</th>\n",
       "      <th>date</th>\n",
       "      <th>active_vehicles</th>\n",
       "      <th>trips</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>B02512</td>\n",
       "      <td>1/1/2015</td>\n",
       "      <td>190</td>\n",
       "      <td>1132</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>B02765</td>\n",
       "      <td>1/1/2015</td>\n",
       "      <td>225</td>\n",
       "      <td>1765</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>B02764</td>\n",
       "      <td>1/1/2015</td>\n",
       "      <td>3427</td>\n",
       "      <td>29421</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>B02682</td>\n",
       "      <td>1/1/2015</td>\n",
       "      <td>945</td>\n",
       "      <td>7679</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>B02617</td>\n",
       "      <td>1/1/2015</td>\n",
       "      <td>1228</td>\n",
       "      <td>9537</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>B02598</td>\n",
       "      <td>1/1/2015</td>\n",
       "      <td>870</td>\n",
       "      <td>6903</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>B02598</td>\n",
       "      <td>1/2/2015</td>\n",
       "      <td>785</td>\n",
       "      <td>4768</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>B02617</td>\n",
       "      <td>1/2/2015</td>\n",
       "      <td>1137</td>\n",
       "      <td>7065</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>B02512</td>\n",
       "      <td>1/2/2015</td>\n",
       "      <td>175</td>\n",
       "      <td>875</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>B02682</td>\n",
       "      <td>1/2/2015</td>\n",
       "      <td>890</td>\n",
       "      <td>5506</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  dispatching_base_number      date  active_vehicles  trips\n",
       "0                  B02512  1/1/2015              190   1132\n",
       "1                  B02765  1/1/2015              225   1765\n",
       "2                  B02764  1/1/2015             3427  29421\n",
       "3                  B02682  1/1/2015              945   7679\n",
       "4                  B02617  1/1/2015             1228   9537\n",
       "5                  B02598  1/1/2015              870   6903\n",
       "6                  B02598  1/2/2015              785   4768\n",
       "7                  B02617  1/2/2015             1137   7065\n",
       "8                  B02512  1/2/2015              175    875\n",
       "9                  B02682  1/2/2015              890   5506"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Visualizando as primeiras linhas\n",
    "uberFile.head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformando DataFrame Pandas em Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Tranformando o dataframe (Pandas) em um Dataframe (Spark)\n",
    "uberDF = sqlContext.createDataFrame(uberFile)\n",
    "# createDataFrame do sqlContext, passando como parâmetro o conjunto Pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tipo DataFrame pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(uberDF)\n",
    "# Tipo DataFrame Pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Passando arquivo .csv no Spark sc.textFile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando o RDD a partir do arquivo csv\n",
    "uberRDD = sc.textFile(\"data/uber.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tipo RDD pyspark - RDD devidamente criado com sc.textFile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.RDD"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(uberRDD)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "355"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Total de registros\n",
    "uberRDD.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Retornando cabeçalho do RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'dispatching_base_number,date,active_vehicles,trips'"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Primeiro registro\n",
    "uberRDD.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dividir conjunto de dados em colunas\n",
    "\n",
    "- ler o arquivo com sc.textFile retornando o arquivo RDD\n",
    "- função map() aplicada ao uberRDD criado acima.\n",
    "- função anônima lambda que receberá cada \"line\" do conjunto de dados\n",
    "- lambda devolve essa \"line\" dividida por virgula - line.split(\",\")\n",
    "- output: colunas são separadas por linhas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dividindo o arquivo em colunas, separadas pelo caracter \",\"\"\n",
    "uberLinhas = uberRDD.map(lambda line: line.split(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tipo Pipeline RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.PipelinedRDD"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(uberLinhas)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Contabilizar número de bases Uber\n",
    "\n",
    "- Função map() para realizar uma TRANSFORMAÇÃO \n",
    "- Aplicar a função mapa() ao PipelineRDD \"uberLinhas\"\n",
    "- Dentro de map() temos a função anônima lambda\n",
    "- lambda recebe linhas como entrada e retorna linha no índice 0\n",
    "- Retornar todas as linhas Índice 0 - \"dispatching_base_number\n",
    "- Retornar todas as linhas distintas .distinct()\n",
    "- Realizar a contagem das linahs distintas .count() \n",
    "- -1 para remover o cabeçalho da contagem das linhas do índice\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Temos 6 bases Uber distintas "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Número de bases de carros do Uber\n",
    "uberLinhas.map(lambda linha: linha[0]).distinct().count() -1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Retornar apenas o Nome das bases\n",
    "- Aplicar a mesma regra anterior ao pipeLine RDD\n",
    "- Ao invés de passar o count(), passar collect()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['dispatching_base_number',\n",
       " 'B02765',\n",
       " 'B02682',\n",
       " 'B02598',\n",
       " 'B02512',\n",
       " 'B02764',\n",
       " 'B02617']"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Bases de carros do Uber\n",
    "uberLinhas.map(lambda linha: linha[0]).distinct().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aplicar filtro para contar carros de B02617\n",
    "\n",
    "- Aplicar filtro ao pipeLine RDD\n",
    "- Filtro recebe a função anônima lambda\n",
    "- Lambda recebe todas as linhas como entrada \n",
    "- Retornar apenas \"B02617\" nas linhas\n",
    "- Realizar contagem de linhas com o filtro estabelecido\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "59"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Total de veículos que passaram pela base B02617\n",
    "uberLinhas.filter(lambda linha: \"B02617\" in linha).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Gravar a informação das bases em um RDD\n",
    "\n",
    "- gravar o comando dentro do RDD - b02617_RDD\n",
    "- passando a mesma função já com o resultado sem apicar .count()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Gravando os dados dos veículos da base B02617 em um novo RDD\n",
    "b02617_RDD = uberLinhas.filter(lambda linha: \"B02617\" in linha)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Filtras dias que o número de corridas > 16.000\n",
    "\n",
    "- Aplicar filter ao RDD salvo b02617\n",
    "- Passar como parâmetro a função lambda\n",
    "- Para cada (linha:), retornar a linha dentro do Índice 3\n",
    "- Aonde o valor do Índice \"trips\" seja maior que 16.000\n",
    "- Fazer a contagem dessas linhas .count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Total de dias em que o número de corridas foi superior a 16.000\n",
    "b02617_RDD.filter(lambda linha: int(linha[3]) > 16000).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Quais os dias com número de viagens > 16.000\n",
    "\n",
    "- Aplicar a mesma regra ao RDD \n",
    "- Trocar o count() por collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['B02617', '2/13/2015', '1590', '16996'],\n",
       " ['B02617', '2/14/2015', '1486', '16999'],\n",
       " ['B02617', '2/20/2015', '1574', '16856'],\n",
       " ['B02617', '2/21/2015', '1443', '16098']]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Dias em que o total de corridas foi superior a 16.000\n",
    "b02617_RDD.filter(lambda linha: int(linha[3]) > 16000).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Criar novo RDD a partir do arquivo original\n",
    "\n",
    "- ler o arquivo .csv através de sc.textFile\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando um novo RDD\n",
    "uberRDD2 = sc.textFile(\"data/uber.csv\").filter(lambda line: \"base\" not in line).map(lambda line:line.split(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Realizar Redução\n",
    "\n",
    "## Qual o número total de corridas por base\n",
    "\n",
    "- Aplicar a função map() ao RDD\n",
    "- Passar como parâmetro a função lambda\n",
    "- Colocar os dados em pares de chave e valor\n",
    "- Fazer a redução por chame\n",
    "- Gerar collect \n",
    "- Aplicar takeOrderes para ordenar "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('B02765', 193670),\n",
       " ('B02682', 662509),\n",
       " ('B02598', 540791),\n",
       " ('B02512', 93786),\n",
       " ('B02764', 1914449),\n",
       " ('B02617', 725025)]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Aplicando redução para calcular o total por base\n",
    "uberRDD2.map(lambda kp: (kp[0], int(kp[3])) ).reduceByKey(lambda k,v: k + v).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('B02764', 1914449),\n",
       " ('B02617', 725025),\n",
       " ('B02682', 662509),\n",
       " ('B02598', 540791),\n",
       " ('B02765', 193670),\n",
       " ('B02512', 93786)]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Aplicando redução para calcular o total por base, em ordem decrescente\n",
    "uberRDD2.map(lambda kp: (kp[0], int(kp[3])) ).reduceByKey(lambda k,v: k + v).takeOrdered(10, key = lambda x: -x[1])"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}