Spark-SQL.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 8</font>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img width='400px' src='spark.png'>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### *********** Atenção: *********** \n",
    "Utilize Java JDK 1.8 e Apache Spark 2.4.2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Java JDK 1.8:\n",
    "\n",
    "https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html"
   ]
  },
  {
   "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": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark SQL - Manipulação de GRANDES conjuntos de dados"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "O Spark SQL é usado para acessar, consultar e manipular dados estruturados com Spark."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img width='700px' src='dataframe.png'>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Acessar http://localhost:4040 para execução dos jobs - User Interface\n",
    "#### Pacotes adicionais podem ser encontrados aqui: https://spark-packages.org/ (usaremos um destes pacotes para conexão com o MongoDB)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark SQL - Spark Session e SQL Context"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# val sqlContext = new org.apache.spark.sql.SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pacote SQL do PySpark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql import Row"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ao executar a linha de comando PySpark, automaticamente é gerado um \"SparkContext\", estabelecendo a conexão com a infraestura do Apache Spark. \n",
    "\n",
    "Neste caso é uma conexão local, a minha própria máquina. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<SparkContext master=local[*] appName=PySparkShell>\n"
     ]
    }
   ],
   "source": [
    "print(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Para trabalhar com Spark SQL, são necessários 3 componentes:\n",
    "\n",
    "### 1. Spark Session.builder.master(\"local\")\n",
    "### 2. SQL Context\n",
    "### 3. DataFrame Estrutura de Dados"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Método .getOrCreate() - Se essa sessão já existir, conectamos. Caso contráro, criamos. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Spark Session - usado para trabalhar com o Spark\n",
    "spSession = SparkSession.builder.master(\"local\").appName(\"Leo-SparkSQL\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Com o SparkSession criado, precisamos de um SQLContext, apenas chamando a função e passando (sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando o SQL Context para trabalhar com Spark SQL\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Criar RDD passando o arquivo .csv em sc.textFile()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importando o arquivo e criando um RDD\n",
    "linhasRDD1 = sc.textFile(\"data/carros.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "198"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "linhasRDD1.count() # Chamar .count() AÇÃO"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Executar uma TRANSFORMAÇÃO - lazy evaluation, fica na fila\n",
    "\n",
    "- Aplicando filtro ao RDD\n",
    "- Passando função anônima como parâmetro\n",
    "- Neste filtro lambda retorna x: sempre que a palavra \"FUELTYPE\" não aparecer em x. \n",
    "- \"FUELTYPE\" é uma das palavras que aparece no dataset\n",
    "\n",
    "## Transformação 1- Filtro not in"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Removendo a primeira linha - Transformação 1\n",
    "linhasRDD2 = linhasRDD1.filter(lambda x: \"FUELTYPE\" not in x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Processada ao chamar a AÇÃO .count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "197"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "linhasRDD2.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformação 2 - Separar por Colunas\n",
    "\n",
    "- Aplicando a função map() ao RDD\n",
    "- Passando como parâmetro a função anônima\n",
    "- Lambda retorna cada linha, para cada linha realizar .split() com caracter \",\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dividindo o conjunto de dados em colunas - Transformação 2\n",
    "linhasRDD3 = linhasRDD2.map(lambda line: line.split(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformação 3 - \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dividindo o conjunto de dados em colunas - Transformação 3\n",
    "linhasRDD4 = linhasRDD3.map(lambda p: Row(make = p[0], body = p[4], hp = int(p[7])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PythonRDD[4] at RDD at PythonRDD.scala:53\n"
     ]
    }
   ],
   "source": [
    "print(linhasRDD4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "?Row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(body='hatchback', hp=69, make='subaru'),\n",
       " Row(body='hatchback', hp=48, make='chevrolet'),\n",
       " Row(body='hatchback', hp=68, make='mazda'),\n",
       " Row(body='hatchback', hp=62, make='toyota'),\n",
       " Row(body='hatchback', hp=68, make='mitsubishi'),\n",
       " Row(body='hatchback', hp=60, make='honda'),\n",
       " Row(body='sedan', hp=69, make='nissan'),\n",
       " Row(body='hatchback', hp=68, make='dodge'),\n",
       " Row(body='hatchback', hp=68, make='plymouth'),\n",
       " Row(body='hatchback', hp=68, make='mazda'),\n",
       " Row(body='hatchback', hp=68, make='mitsubishi'),\n",
       " Row(body='hatchback', hp=68, make='dodge'),\n",
       " Row(body='hatchback', hp=68, make='plymouth'),\n",
       " Row(body='hatchback', hp=70, make='chevrolet'),\n",
       " Row(body='hatchback', hp=62, make='toyota'),\n",
       " Row(body='hatchback', hp=68, make='dodge'),\n",
       " Row(body='hatchback', hp=58, make='honda'),\n",
       " Row(body='hatchback', hp=62, make='toyota'),\n",
       " Row(body='hatchback', hp=76, make='honda'),\n",
       " Row(body='sedan', hp=70, make='chevrolet'),\n",
       " Row(body='sedan', hp=69, make='nissan'),\n",
       " Row(body='hatchback', hp=68, make='mitsubishi'),\n",
       " Row(body='sedan', hp=68, make='dodge'),\n",
       " Row(body='sedan', hp=68, make='plymouth'),\n",
       " Row(body='sedan', hp=68, make='mazda'),\n",
       " Row(body='sedan', hp=78, make='isuzu'),\n",
       " Row(body='hatchback', hp=68, make='mazda'),\n",
       " Row(body='sedan', hp=69, make='nissan'),\n",
       " Row(body='hatchback', hp=76, make='honda'),\n",
       " Row(body='wagon', hp=62, make='toyota'),\n",
       " Row(body='sedan', hp=70, make='toyota'),\n",
       " Row(body='sedan', hp=88, make='mitsubishi'),\n",
       " Row(body='hatchback', hp=73, make='subaru'),\n",
       " Row(body='sedan', hp=55, make='nissan'),\n",
       " Row(body='sedan', hp=82, make='subaru'),\n",
       " Row(body='hatchback', hp=76, make='honda'),\n",
       " Row(body='hatchback', hp=70, make='toyota'),\n",
       " Row(body='sedan', hp=76, make='honda'),\n",
       " Row(body='wagon', hp=76, make='honda'),\n",
       " Row(body='sedan', hp=69, make='nissan'),\n",
       " Row(body='wagon', hp=69, make='nissan'),\n",
       " Row(body='sedan', hp=68, make='mazda'),\n",
       " Row(body='wagon', hp=82, make='subaru'),\n",
       " Row(body='sedan', hp=69, make='nissan'),\n",
       " Row(body='hatchback', hp=73, make='subaru'),\n",
       " Row(body='sedan', hp=68, make='dodge'),\n",
       " Row(body='sedan', hp=68, make='plymouth'),\n",
       " Row(body='hatchback', hp=102, make='mitsubishi'),\n",
       " Row(body='sedan', hp=70, make='toyota'),\n",
       " Row(body='sedan', hp=82, make='subaru'),\n",
       " Row(body='sedan', hp=52, make='volkswagen'),\n",
       " Row(body='hatchback', hp=56, make='toyota'),\n",
       " Row(body='hatchback', hp=69, make='nissan'),\n",
       " Row(body='hatchback', hp=86, make='honda'),\n",
       " Row(body='wagon', hp=62, make='toyota'),\n",
       " Row(body='sedan', hp=56, make='toyota'),\n",
       " Row(body='hatchback', hp=102, make='dodge'),\n",
       " Row(body='hatchback', hp=102, make='plymouth'),\n",
       " Row(body='sedan', hp=85, make='volkswagen'),\n",
       " Row(body='sedan', hp=52, make='volkswagen'),\n",
       " Row(body='wagon', hp=69, make='nissan'),\n",
       " Row(body='wagon', hp=82, make='subaru'),\n",
       " Row(body='sedan', hp=70, make='toyota'),\n",
       " Row(body='sedan', hp=88, make='mitsubishi'),\n",
       " Row(body='sedan', hp=85, make='volkswagen'),\n",
       " Row(body='hatchback', hp=70, make='toyota'),\n",
       " Row(body='hardtop', hp=69, make='nissan'),\n",
       " Row(body='hatchback', hp=70, make='toyota'),\n",
       " Row(body='hardtop', hp=116, make='toyota'),\n",
       " Row(body='sedan', hp=84, make='mazda'),\n",
       " Row(body='sedan', hp=85, make='volkswagen'),\n",
       " Row(body='hatchback', hp=88, make='mitsubishi'),\n",
       " Row(body='wagon', hp=62, make='toyota'),\n",
       " Row(body='sedan', hp=86, make='honda'),\n",
       " Row(body='hatchback', hp=84, make='mazda'),\n",
       " Row(body='wagon', hp=88, make='dodge'),\n",
       " Row(body='wagon', hp=88, make='plymouth'),\n",
       " Row(body='sedan', hp=92, make='toyota'),\n",
       " Row(body='hatchback', hp=97, make='nissan'),\n",
       " Row(body='hatchback', hp=86, make='honda'),\n",
       " Row(body='sedan', hp=82, make='subaru'),\n",
       " Row(body='sedan', hp=70, make='toyota'),\n",
       " Row(body='sedan', hp=116, make='mitsubishi'),\n",
       " Row(body='sedan', hp=116, make='mitsubishi'),\n",
       " Row(body='sedan', hp=112, make='toyota'),\n",
       " Row(body='sedan', hp=68, make='volkswagen'),\n",
       " Row(body='hatchback', hp=112, make='toyota'),\n",
       " Row(body='sedan', hp=97, make='nissan'),\n",
       " Row(body='hardtop', hp=116, make='toyota'),\n",
       " Row(body='hatchback', hp=116, make='mitsubishi'),\n",
       " Row(body='sedan', hp=94, make='subaru'),\n",
       " Row(body='hatchback', hp=90, make='volkswagen'),\n",
       " Row(body='hatchback', hp=92, make='toyota'),\n",
       " Row(body='hatchback', hp=116, make='toyota'),\n",
       " Row(body='sedan', hp=100, make='volkswagen'),\n",
       " Row(body='wagon', hp=94, make='subaru'),\n",
       " Row(body='sedan', hp=84, make='mazda'),\n",
       " Row(body='sedan', hp=86, make='honda'),\n",
       " Row(body='sedan', hp=100, make='honda'),\n",
       " Row(body='hatchback', hp=84, make='mazda'),\n",
       " Row(body='sedan', hp=73, make='toyota'),\n",
       " Row(body='sedan', hp=92, make='toyota'),\n",
       " Row(body='hatchback', hp=101, make='mazda'),\n",
       " Row(body='hatchback', hp=90, make='isuzu'),\n",
       " Row(body='hardtop', hp=116, make='toyota'),\n",
       " Row(body='hatchback', hp=84, make='mazda'),\n",
       " Row(body='hatchback', hp=92, make='toyota'),\n",
       " Row(body='sedan', hp=111, make='subaru'),\n",
       " Row(body='hatchback', hp=116, make='toyota'),\n",
       " Row(body='convertible', hp=90, make='volkswagen'),\n",
       " Row(body='wagon', hp=111, make='subaru'),\n",
       " Row(body='hatchback', hp=101, make='mazda'),\n",
       " Row(body='hatchback', hp=110, make='saab'),\n",
       " Row(body='sedan', hp=97, make='peugot'),\n",
       " Row(body='sedan', hp=110, make='saab'),\n",
       " Row(body='wagon', hp=88, make='volkswagen'),\n",
       " Row(body='wagon', hp=97, make='peugot'),\n",
       " Row(body='hatchback', hp=145, make='mitsubishi'),\n",
       " Row(body='hatchback', hp=145, make='plymouth'),\n",
       " Row(body='sedan', hp=114, make='volvo'),\n",
       " Row(body='sedan', hp=101, make='honda'),\n",
       " Row(body='hatchback', hp=145, make='dodge'),\n",
       " Row(body='sedan', hp=95, make='peugot'),\n",
       " Row(body='sedan', hp=110, make='volkswagen'),\n",
       " Row(body='wagon', hp=114, make='volvo'),\n",
       " Row(body='convertible', hp=111, make='alfa-romero'),\n",
       " Row(body='sedan', hp=152, make='nissan'),\n",
       " Row(body='sedan', hp=152, make='nissan'),\n",
       " Row(body='hatchback', hp=101, make='mazda'),\n",
       " Row(body='sedan', hp=68, make='volkswagen'),\n",
       " Row(body='wagon', hp=95, make='peugot'),\n",
       " Row(body='sedan', hp=102, make='audi'),\n",
       " Row(body='wagon', hp=152, make='nissan'),\n",
       " Row(body='hatchback', hp=145, make='mitsubishi'),\n",
       " Row(body='hatchback', hp=145, make='mitsubishi'),\n",
       " Row(body='hatchback', hp=110, make='saab'),\n",
       " Row(body='sedan', hp=110, make='audi'),\n",
       " Row(body='sedan', hp=110, make='saab'),\n",
       " Row(body='sedan', hp=95, make='peugot'),\n",
       " Row(body='hatchback', hp=135, make='mazda'),\n",
       " Row(body='sedan', hp=156, make='toyota'),\n",
       " Row(body='wagon', hp=156, make='toyota'),\n",
       " Row(body='sedan', hp=114, make='volvo'),\n",
       " Row(body='hatchback', hp=161, make='toyota'),\n",
       " Row(body='sedan', hp=101, make='bmw'),\n",
       " Row(body='convertible', hp=111, make='alfa-romero'),\n",
       " Row(body='hatchback', hp=154, make='alfa-romero'),\n",
       " Row(body='hatchback', hp=175, make='mercury'),\n",
       " Row(body='wagon', hp=114, make='volvo'),\n",
       " Row(body='hatchback', hp=161, make='toyota'),\n",
       " Row(body='sedan', hp=97, make='peugot'),\n",
       " Row(body='wagon', hp=95, make='peugot'),\n",
       " Row(body='sedan', hp=114, make='volvo'),\n",
       " Row(body='sedan', hp=95, make='peugot'),\n",
       " Row(body='sedan', hp=101, make='bmw'),\n",
       " Row(body='wagon', hp=95, make='peugot'),\n",
       " Row(body='hatchback', hp=160, make='nissan'),\n",
       " Row(body='sedan', hp=115, make='audi'),\n",
       " Row(body='convertible', hp=116, make='toyota'),\n",
       " Row(body='sedan', hp=110, make='audi'),\n",
       " Row(body='sedan', hp=95, make='peugot'),\n",
       " Row(body='sedan', hp=142, make='peugot'),\n",
       " Row(body='hatchback', hp=160, make='saab'),\n",
       " Row(body='sedan', hp=120, make='mazda'),\n",
       " Row(body='sedan', hp=72, make='mazda'),\n",
       " Row(body='hatchback', hp=160, make='nissan'),\n",
       " Row(body='sedan', hp=162, make='volvo'),\n",
       " Row(body='sedan', hp=160, make='saab'),\n",
       " Row(body='wagon', hp=110, make='audi'),\n",
       " Row(body='wagon', hp=162, make='volvo'),\n",
       " Row(body='sedan', hp=160, make='volvo'),\n",
       " Row(body='hatchback', hp=200, make='nissan'),\n",
       " Row(body='sedan', hp=121, make='bmw'),\n",
       " Row(body='sedan', hp=121, make='bmw'),\n",
       " Row(body='sedan', hp=134, make='volvo'),\n",
       " Row(body='hatchback', hp=143, make='porsche'),\n",
       " Row(body='sedan', hp=106, make='volvo'),\n",
       " Row(body='sedan', hp=114, make='volvo'),\n",
       " Row(body='sedan', hp=140, make='audi'),\n",
       " Row(body='sedan', hp=121, make='bmw'),\n",
       " Row(body='sedan', hp=123, make='mercedes-benz'),\n",
       " Row(body='hardtop', hp=123, make='mercedes-benz'),\n",
       " Row(body='wagon', hp=123, make='mercedes-benz'),\n",
       " Row(body='sedan', hp=182, make='bmw'),\n",
       " Row(body='sedan', hp=123, make='mercedes-benz'),\n",
       " Row(body='sedan', hp=176, make='jaguar'),\n",
       " Row(body='hardtop', hp=207, make='porsche'),\n",
       " Row(body='hardtop', hp=207, make='porsche'),\n",
       " Row(body='sedan', hp=155, make='mercedes-benz'),\n",
       " Row(body='convertible', hp=155, make='mercedes-benz'),\n",
       " Row(body='sedan', hp=176, make='jaguar'),\n",
       " Row(body='sedan', hp=262, make='jaguar'),\n",
       " Row(body='sedan', hp=182, make='bmw'),\n",
       " Row(body='convertible', hp=207, make='porsche'),\n",
       " Row(body='sedan', hp=184, make='mercedes-benz'),\n",
       " Row(body='sedan', hp=182, make='bmw'),\n",
       " Row(body='hardtop', hp=184, make='mercedes-benz')]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "linhasRDD4.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Criar Tabela|DataFrame a partir do spSession passando como parâmetro um RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando um dataframe a partir do RDD\n",
    "linhasDF = spSession.createDataFrame(linhasRDD4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---+----------+\n",
      "|     body| hp|      make|\n",
      "+---------+---+----------+\n",
      "|hatchback| 69|    subaru|\n",
      "|hatchback| 48| chevrolet|\n",
      "|hatchback| 68|     mazda|\n",
      "|hatchback| 62|    toyota|\n",
      "|hatchback| 68|mitsubishi|\n",
      "|hatchback| 60|     honda|\n",
      "|    sedan| 69|    nissan|\n",
      "|hatchback| 68|     dodge|\n",
      "|hatchback| 68|  plymouth|\n",
      "|hatchback| 68|     mazda|\n",
      "|hatchback| 68|mitsubishi|\n",
      "|hatchback| 68|     dodge|\n",
      "|hatchback| 68|  plymouth|\n",
      "|hatchback| 70| chevrolet|\n",
      "|hatchback| 62|    toyota|\n",
      "|hatchback| 68|     dodge|\n",
      "|hatchback| 58|     honda|\n",
      "|hatchback| 62|    toyota|\n",
      "|hatchback| 76|     honda|\n",
      "|    sedan| 70| chevrolet|\n",
      "+---------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "linhasDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(linhasDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---+----------+\n",
      "|     body| hp|      make|\n",
      "+---------+---+----------+\n",
      "|hatchback| 69|    subaru|\n",
      "|hatchback| 48| chevrolet|\n",
      "|hatchback| 68|     mazda|\n",
      "|hatchback| 62|    toyota|\n",
      "|hatchback| 68|mitsubishi|\n",
      "|hatchback| 60|     honda|\n",
      "|    sedan| 69|    nissan|\n",
      "|hatchback| 68|     dodge|\n",
      "|hatchback| 68|  plymouth|\n",
      "|hatchback| 68|     mazda|\n",
      "|hatchback| 68|mitsubishi|\n",
      "|hatchback| 68|     dodge|\n",
      "|hatchback| 68|  plymouth|\n",
      "|hatchback| 70| chevrolet|\n",
      "|hatchback| 62|    toyota|\n",
      "|hatchback| 68|     dodge|\n",
      "|hatchback| 58|     honda|\n",
      "|hatchback| 62|    toyota|\n",
      "|hatchback| 76|     honda|\n",
      "|    sedan| 70| chevrolet|\n",
      "+---------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Mesma coisa que: SELECT * FROM linhasDF\n",
    "linhasDF.select(\"*\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---+-----------+\n",
      "|       body| hp|       make|\n",
      "+-----------+---+-----------+\n",
      "|  hatchback|154|alfa-romero|\n",
      "|convertible|111|alfa-romero|\n",
      "|convertible|111|alfa-romero|\n",
      "|      sedan|110|       audi|\n",
      "|      sedan|115|       audi|\n",
      "|      sedan|110|       audi|\n",
      "|      wagon|110|       audi|\n",
      "|      sedan|140|       audi|\n",
      "|      sedan|102|       audi|\n",
      "|      sedan|101|        bmw|\n",
      "|      sedan|101|        bmw|\n",
      "|      sedan|121|        bmw|\n",
      "|      sedan|121|        bmw|\n",
      "|      sedan|182|        bmw|\n",
      "|      sedan|182|        bmw|\n",
      "|      sedan|121|        bmw|\n",
      "|      sedan|182|        bmw|\n",
      "|      sedan| 70|  chevrolet|\n",
      "|  hatchback| 70|  chevrolet|\n",
      "|  hatchback| 48|  chevrolet|\n",
      "+-----------+---+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Mesma coisa que: SELECT * FROM linhasDF ORDER BY make\n",
    "linhasDF.orderBy(\"make\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Criando tabela temporária"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Registrando o dataframe como uma Temp Table\n",
    "linhasDF.createOrReplaceTempView(\"linhasTB\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "java version \"11.0.6\" 2020-01-14 LTS\n",
      "Java(TM) SE Runtime Environment 18.9 (build 11.0.6+8-LTS)\n",
      "Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.6+8-LTS, mixed mode)\n"
     ]
    }
   ],
   "source": [
    "!java -version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Executando queries SQL ANSI --- JAVA 8\n",
    "spSession.sql(\"select * from linhasTB where make = 'nissan'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Executando queries SQL ANSI ---- JAVA 8\n",
    "spSession.sql(\"select make, body, avg(hp) from linhasTB group by make, body\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark SQL e Arquivos CSV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "carrosDF = spSession.read.csv(\"data/carros.csv\", header = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(carrosDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+------+-----+---------+-----+---------+---+----+--------+-------+-----+\n",
      "|      MAKE|FUELTYPE|ASPIRE|DOORS|     BODY|DRIVE|CYLINDERS| HP| RPM|MPG-CITY|MPG-HWY|PRICE|\n",
      "+----------+--------+------+-----+---------+-----+---------+---+----+--------+-------+-----+\n",
      "|    subaru|     gas|   std|  two|hatchback|  fwd|     four| 69|4900|      31|     36| 5118|\n",
      "| chevrolet|     gas|   std|  two|hatchback|  fwd|    three| 48|5100|      47|     53| 5151|\n",
      "|     mazda|     gas|   std|  two|hatchback|  fwd|     four| 68|5000|      30|     31| 5195|\n",
      "|    toyota|     gas|   std|  two|hatchback|  fwd|     four| 62|4800|      35|     39| 5348|\n",
      "|mitsubishi|     gas|   std|  two|hatchback|  fwd|     four| 68|5500|      37|     41| 5389|\n",
      "|     honda|     gas|   std|  two|hatchback|  fwd|     four| 60|5500|      38|     42| 5399|\n",
      "|    nissan|     gas|   std|  two|    sedan|  fwd|     four| 69|5200|      31|     37| 5499|\n",
      "|     dodge|     gas|   std|  two|hatchback|  fwd|     four| 68|5500|      37|     41| 5572|\n",
      "|  plymouth|     gas|   std|  two|hatchback|  fwd|     four| 68|5500|      37|     41| 5572|\n",
      "|     mazda|     gas|   std|  two|hatchback|  fwd|     four| 68|5000|      31|     38| 6095|\n",
      "|mitsubishi|     gas|   std|  two|hatchback|  fwd|     four| 68|5500|      31|     38| 6189|\n",
      "|     dodge|     gas|   std| four|hatchback|  fwd|     four| 68|5500|      31|     38| 6229|\n",
      "|  plymouth|     gas|   std| four|hatchback|  fwd|     four| 68|5500|      31|     38| 6229|\n",
      "| chevrolet|     gas|   std|  two|hatchback|  fwd|     four| 70|5400|      38|     43| 6295|\n",
      "|    toyota|     gas|   std|  two|hatchback|  fwd|     four| 62|4800|      31|     38| 6338|\n",
      "|     dodge|     gas|   std|  two|hatchback|  fwd|     four| 68|5500|      31|     38| 6377|\n",
      "|     honda|     gas|   std|  two|hatchback|  fwd|     four| 58|4800|      49|     54| 6479|\n",
      "|    toyota|     gas|   std| four|hatchback|  fwd|     four| 62|4800|      31|     38| 6488|\n",
      "|     honda|     gas|   std|  two|hatchback|  fwd|     four| 76|6000|      30|     34| 6529|\n",
      "| chevrolet|     gas|   std| four|    sedan|  fwd|     four| 70|5400|      38|     43| 6575|\n",
      "+----------+--------+------+-----+---------+-----+---------+---+----+--------+-------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "carrosDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Registrando o dataframe como uma Temp Table\n",
    "carrosDF.createOrReplaceTempView(\"carrosTB\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Executando queries SQL ANSI\n",
    "spSession.sql(\"select make, hp, price from carrosTB where CYLINDERS = 'three'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "carrosTT = spSession.sql(\"select make, hp, price from carrosTB where CYLINDERS = 'three'\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "carrosTT.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aplicando Machine Learning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "data/carros.csv MapPartitionsRDD[40] at textFile at NativeMethodAccessorImpl.java:0"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Carregando o arquivo CSV e mantendo o objeto em cache\n",
    "carros = sc.textFile(\"data/carros.csv\")\n",
    "carros.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "197"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Remove a primeira linha (header)\n",
    "primeiraLinha = carros.first()\n",
    "linhas = carros.filter(lambda x: x != primeiraLinha)\n",
    "linhas.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importando função row\n",
    "from pyspark.sql import Row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convertendo para um vetor de linhas\n",
    "def transformToNumeric(inputStr) :\n",
    "    \n",
    "    attList = inputStr.split(\",\")\n",
    "    \n",
    "    doors = 1.0 if attList[3] == \"two\" else 2.0\n",
    "    \n",
    "    body = 1.0 if attList[4] == \"sedan\" else 2.0 \n",
    "       \n",
    "    # Filtrando colunas não necessárias nesta etapa\n",
    "    valores = Row(DOORS = doors, BODY = float(body), HP = float(attList[7]), RPM = float(attList[8]), MPG = float(attList[9]))\n",
    "    return valores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Aplicando a função aos dados e persistindo o resultado em memória\n",
    "autoMap = linhas.map(transformToNumeric)\n",
    "autoMap.persist()\n",
    "autoMap.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando o Dataframe\n",
    "carrosDf = spSession.createDataFrame(autoMap)\n",
    "carrosDf.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## .decribe() Descrever estatísticas do DF, estruturar em Pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sumarizando as estatísticas do conjunto de dados\n",
    "summStats = carrosDf.describe().toPandas()\n",
    "summStats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extraindo as médias\n",
    "medias = summStats.iloc[1,1:5].values.tolist()\n",
    "medias"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extraindo o desvio padrão\n",
    "desvios_padroes = summStats.iloc[2,1:5].values.tolist()\n",
    "desvios_padroes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Inserindo a média e o desvio padrão em uma variável do tipo broadcast \n",
    "bcMedias = sc.broadcast(medias)\n",
    "bcDesviosP = sc.broadcast(desvios_padroes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importando a Função Vectors\n",
    "from pyspark.ml.linalg import Vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Função para normalizar os dados e criar um vetor denso\n",
    "def centerAndScale(inRow) :\n",
    "    global bcMedias\n",
    "    global bcDesviosP\n",
    "    \n",
    "    meanArray = bcMedias.value\n",
    "    stdArray = bcDesviosP.value\n",
    "\n",
    "    retArray = [] # array vazio \n",
    "    \n",
    "# Para cada valor de i no range de valores, no comprimento de array de médias    \n",
    "    for i in range(len(meanArray)):\n",
    "# Pegar o Array vazio e adicionar a narmalização para deixar os dados na mesma escala\n",
    "        retArray.append( (float(inRow[i]) - float(meanArray[i])) / float(stdArray[i]) )\n",
    "# \n",
    "    return Vectors.dense(retArray)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Aplicando a normalização aos dados\n",
    "csAuto = carrosDf.rdd.map(centerAndScale)\n",
    "csAuto.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando um Spark Dataframe com as features (atributos)\n",
    "autoRows = csAuto.map(lambda f: Row(features = f))\n",
    "autoDf = spSession.createDataFrame(autoRows)\n",
    "autoDf.select(\"features\").show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importando o algoritmo K-Means para clusterização\n",
    "from pyspark.ml.clustering import KMeans\n",
    "kmeans = KMeans(k = 3, seed = 1)\n",
    "modelo = kmeans.fit(autoDf)\n",
    "previsoes = modelo.transform(autoDf)\n",
    "previsoes.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Plot dos resultados\n",
    "import pandas as pd\n",
    "import matplotlib.pylab as plt\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Função para leitura dos dados e plotagem\n",
    "def unstripData(instr) :\n",
    "    return ( instr[\"prediction\"], instr[\"features\"][0], instr[\"features\"][1],instr[\"features\"][2],instr[\"features\"][3])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Organizando os dados para o Plot\n",
    "unstripped = previsoes.rdd.map(unstripData)\n",
    "predList = unstripped.collect()\n",
    "predPd = pd.DataFrame(predList)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.cla()\n",
    "plt.scatter(predPd[3], predPd[4], c = predPd[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Spark SQL e Arquivos JSON"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Neste site você pode validar a estrutura de um arquivo JSON: http://jsonlint.com/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importando o arquivo JSON\n",
    "funcDF = spSession.read.json(\"data/funcionarios.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "funcDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "funcDF.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(funcDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Operações com Dataframe Spark SQL - select()\n",
    "funcDF.select(\"nome\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Operações com Dataframe Spark SQL - filter()\n",
    "funcDF.filter(funcDF[\"idade\"] == 50).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Operações com Dataframe Spark SQL - groupBy()\n",
    "funcDF.groupBy(\"sexo\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Operações com Dataframe Spark SQL - groupBy()\n",
    "funcDF.groupBy(\"deptid\").agg({\"salario\": \"avg\", \"idade\": \"max\"}).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Registrando o dataframe como uma Temp Table\n",
    "funcDF.registerTempTable(\"funcTB\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Executando queries SQL ANSI\n",
    "spSession.sql(\"select deptid, max(idade), avg(salario) from funcTB group by deptid\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Temp Tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Registrando o dataframe como temp Table\n",
    "funcDF.createOrReplaceTempView(\"funcTB\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spSession.sql(\"select * from funcTB where salario = 9700\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Criando Temp Table\n",
    "sqlContext.registerDataFrameAsTable(funcDF, \"funcTB2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(funcTB2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Persistindo a Temp Table \n",
    "funcTB3 = spSession.table(\"funcTB2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(funcTB3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Comparando o Dataframe com a tabela temporária criada\n",
    "sorted(funcDF.collect()) == sorted(funcTB3.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Aplicando o filtro\n",
    "sqlContext.registerDataFrameAsTable(funcDF, \"funcTB2\")\n",
    "funcTB3 = spSession.table(\"funcTB2\")\n",
    "funcTB3.filter(\"idade = '42'\").first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Drop Temp Table\n",
    "sqlContext.dropTempTable(\"funcTB2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Banco de Dados Relacional"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Extraindo Dados do MySQL. Primeiro precisamos baixar o driver JDBC. Haverá um driver JDBC para cada banco de dados que você conectar (Oracle, SQL Server, etc...)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1- Download do Driver JDBC para o MySQL: http://dev.mysql.com/downloads/connector/j/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "2- Baixar o arquivo .zip"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "3- Descompactar o arquivo e copiar o arquivo mysql-connector-java-8.0.16.jar para a pasta /opt/Spark/jars ou para SO Windows em C:\\Spark\\jars"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import SQLContext\n",
    "spSession = SparkSession.builder.master(\"local\").appName(\"DSA-SparkSQL\").getOrCreate()\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "ename": "Py4JJavaError",
     "evalue": "An error occurred while calling o535.load.\n: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver\r\n\tat java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:471)\r\n\tat java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:588)\r\n\tat java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.DriverRegistry$.register(DriverRegistry.scala:45)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.$anonfun$driverClass$1(JDBCOptions.scala:99)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.$anonfun$driverClass$1$adapted(JDBCOptions.scala:99)\r\n\tat scala.Option.foreach(Option.scala:274)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:99)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:35)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:32)\r\n\tat org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:318)\r\n\tat org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)\r\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)\r\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:167)\r\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:566)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.base/java.lang.Thread.run(Thread.java:834)\r\n",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-53-0cfe9aa2f868>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      5\u001b[0m     \u001b[0mdbtable\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;34m\"carrosTB\"\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      6\u001b[0m     \u001b[0muser\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;34m\"root\"\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 7\u001b[1;33m     password = \"dsa1234\").load()\n\u001b[0m",
      "\u001b[1;32mC:\\spark\\python\\pyspark\\sql\\readwriter.py\u001b[0m in \u001b[0;36mload\u001b[1;34m(self, path, format, schema, **options)\u001b[0m\n\u001b[0;32m    170\u001b[0m             \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_df\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jreader\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mload\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_spark\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPythonUtils\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mtoSeq\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    171\u001b[0m         \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 172\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_df\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jreader\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mload\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    173\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    174\u001b[0m     \u001b[1;33m@\u001b[0m\u001b[0msince\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m1.4\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\spark\\python\\lib\\py4j-0.10.7-src.zip\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1255\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1257\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m   1258\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1259\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\spark\\python\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     61\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mdeco\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     62\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 63\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     64\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     65\u001b[0m             \u001b[0ms\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mjava_exception\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mtoString\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\spark\\python\\lib\\py4j-0.10.7-src.zip\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    326\u001b[0m                 raise Py4JJavaError(\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    330\u001b[0m                 raise Py4JError(\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o535.load.\n: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver\r\n\tat java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:471)\r\n\tat java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:588)\r\n\tat java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.DriverRegistry$.register(DriverRegistry.scala:45)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.$anonfun$driverClass$1(JDBCOptions.scala:99)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.$anonfun$driverClass$1$adapted(JDBCOptions.scala:99)\r\n\tat scala.Option.foreach(Option.scala:274)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:99)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:35)\r\n\tat org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:32)\r\n\tat org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:318)\r\n\tat org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)\r\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)\r\n\tat org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:167)\r\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:566)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.base/java.lang.Thread.run(Thread.java:834)\r\n"
     ]
    }
   ],
   "source": [
    "mysql_df = spSession.read.format(\"jdbc\").options(\n",
    "    url = \"jdbc:mysql://localhost/carros\",\n",
    "    serverTimezone = \"UTC\",\n",
    "    driver = \"com.mysql.jdbc.Driver\",\n",
    "    dbtable = \"carrosTB\",\n",
    "    user = \"root\",\n",
    "    password = \"dsa1234\").load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mysql_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mysql_df.registerTempTable(\"carrostb\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spSession.sql(\"select * from carrostb where hp = '68'\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Banco de Dados Não-Relacional"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark Connector: https://docs.mongodb.com/spark-connector/current/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mongo Spark: https://spark-packages.org/package/mongodb/mongo-spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "$SPARK_HOME/bin/pyspark --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Imports\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Leitura"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Cria a sessão\n",
    "my_spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"myApp\") \\\n",
    "    .config(\"spark.mongodb.input.uri\", \"mongodb://localhost/test_db.test_collection\") \\\n",
    "    .config(\"spark.mongodb.output.uri\", \"mongodb://localhost/test_db.test_collection\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Carrega os dados do MongoDB no Spark\n",
    "dados = spark.read.format(\"com.mongodb.spark.sql.DefaultSource\").load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dados.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dados.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dados.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dados.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Gravação"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "registro = spark.createDataFrame([(\"Camisa T-Shirt\",  50)], [\"item\", \"qty\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "registro.write.format(\"com.mongodb.spark.sql.DefaultSource\").mode(\"append\").save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dados.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img width='621' src='joins.png'>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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
}