Analise Final + Relatorio.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](image_banner.png)\n",
    "\n",
    "---\n",
    "# Relatorio ATP - Frameworks de Big Data \n",
    "### Aluno: \n",
    "\n",
    "Juan Manoel Marinho Nascimento\n",
    "[![](https://i.stack.imgur.com/gVE0j.png)](https://www.linkedin.com/in/juan-marinho-6309ba84/)\n",
    "\n",
    "\n",
    "\n",
    "[![](https://i.stack.imgur.com/tskMh.png)](https://github.com/juanengml/)\n",
    "\n",
    "\n",
    "\n",
    "## Introdução \n",
    "Buscou-se  desenvolver  um trabalho prático  utilizando  o  framework  Apache Spark e a tecnologia Python com a api pyspark, no estudo de sistemas distribuídos em big data dentro do ecosistema hadoop. Esta atividade é  relevante  pois  trata-se de  uma  das  principais  tecnologias do mercado em ascenção para processamento de grandes volumes de dados, visto que grandes do mercado utilizam python e spark para seus projetos de alta performace.\n",
    "\n",
    "\n",
    "## Objetivo\n",
    "\n",
    "Analisar  e compreender como funciona o  armazenamento  e  processamento  de  dados  não estruturados por meio de um jupyter notebook desenvolvido em python com o framework Apache Spark usando como api pyspark.\n",
    "\n",
    "##  Métodos \n",
    "Foi análisado os  repositórios  de  datasets  públicos  com  dados  de  diferentes exemplos, reais e artificiais, dentro das bases data.gov, kaggle e outro. Durante essa analise foi percebido e observado a variedade de dados distrintos para os quais é possivel fazer analises e gerar diferentes insights. No desenvolvimento de uma analise programada em pyspark para a utilização do   Apache Spark no  servidor de   big   data. \n",
    "\n",
    "\n",
    "Foi Utilização diversos   algoritmos implementados em python como filtros para entender como os dados estão estruturados e poder filtras as informações, tambem foi utilizado querys em sql visto que pyspark comporta rodar queries sql e utilização de sub-queries, PS: para mais informações basta olhar a analise a baixo. \n",
    "\n",
    "\n",
    "Essa base de dados mostra as principais tipos de ocorrencias e tambem passa os dados de forma não parametrizadas nas columnas, foi feito um processo de data cleaning, para limpar aa base de dados e junção do nome da columa com a base principa. \n",
    "\n",
    "\n",
    "Outro processo importante foi a utilização de agrupamento de dados por dimensões de Ano, Mes e Dia. Com os dados agrupados foi possivel fazer analises das metricas de media, maximo valor, menor valor, e tambem foi possivel segmentar por tipos de crimes\n",
    "\n",
    "---\n",
    "## Resultados e discussões\n",
    "\n",
    "Procurou-se  conhecer e entender os  diferentes  tipos  de  dados em sua maior quantidade e variedade e  sua  organização, fundamentais para seu processamento e analises. Foi utilizado o Apache Spark no servidor de big data  da  PUC-PR  para  reproduzir  o  exemplo da unidade de aprendizagem: “Aplicações simples utilizando frameworks de big data”. Para  a  atividade,  o  seguinte  fluxo  de  execução  foi  implementado  tendo  em  vista  a computação  distribuída:  criação  de  um  jupyter notebook,  criação  de  um  Spark  Context e Session Builder, criação inicial de uma instancia do objeto spark, aplicação da leitura do arquivo csv na pasta local do usuario juan_manoel, feita as transformações e execução das operações.As seguintes transformações foram executadas: filter, groupby , Função Max, mean e juntamente com as seguintes ações: construção de querys e sql utilizando as possibilidades do spark para isso.\n",
    "\n",
    "## Conclusão\n",
    "\n",
    "Conclui-se que a utilização do Apache Spark proporciona ganhos consideráveis para o processamento de grandes conjuntos de dados de forma paralela e distribuída, facilitando e agilizando o desenvolvimento de aplicações de processamento de dados no contexto de big data.A  tecnologia  Python  por  ser  amplamente utilizada em toda a industria de técnologia dentro das areas de inteligencia artifical, ciencias de dados e big data torna o contexto dos  sistemas  distribuídos, mais comodos para o desenvolvimento de soluções em big data.\n",
    "\n",
    "## Referências\n",
    "Google Cloud Platform. Who is Apache Spark. Visão geral apache spark, 2020 disponível em: \n",
    "\n",
    "https://cloud.google.com/learn/what-is-apache-spark\n",
    "\n",
    "API PySpark - Documentação da API, 2020 Disponivel em:\n",
    "\n",
    "https://spark.apache.org/docs/latest/api/python/index.html\n",
    "\n",
    "Course Big Data with Spark - Curso Datacamp com pyspark, 2020 disponivel em: \n",
    "\n",
    "https://www.datacamp.com/tracks/big-data-with-pyspark\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test Check PySpark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "import os \n",
    "os.environ['PYSPARK_PYTHON'] = '/usr/bin/python3.7'\n",
    "os.environ['PYSPARK_DRIVER_PYTHON'] = '/usr/bin/python3.7'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Analise Exploratoria ATP Dataset\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dicionario de Dados\n",
    "\n",
    "O dicionário de dados deste dataset é declarado a seguir:\n",
    "Campo \t\n",
    "\n",
    "| Campo | DESCRIÇÃO |\n",
    "|-------|-----------|\n",
    "| Dia   |  Dia da ocorrência     |\n",
    "| Mes   |  Mes da ocorrência     |\n",
    "| Ano   |  Ano da ocorrência     |\n",
    "| Bloco |Região da ocorrência  |   Região da ocorrência \n",
    "| Tipo  |  Tipo da ocorrência criminal \n",
    "| Descrição |   Breve descrição da ocorrência \n",
    "| Descrição da localização | Descrição da localização da ocorrência (rua, por exemplo) \n",
    "| Latitude |   Localização da ocorrência \n",
    "| Longitude |  Localização da ocorrência "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"ATP Analise Exploratoria\").config(\"spark.some.config.option\",\"some-value\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# USE APENAS QUANDO NÃO TIVER RODANDO NO SERVIDOR LOCAL\n",
    "#!wget -c  https://gitlab.com/juanmanoel/dataset-atp-big-data-frameworks/-/raw/master/ocorrencias_criminais_corrigida.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.option(\"header\",True).csv(\"/home2/ead2020/SEM2/juan.manoel/ocorrencias_criminais_corrigida.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Analise Dataframe entendimento da tabela"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+----+--------------------+--------------------+--------------------+------------------------+------------+-------------+\n",
      "|Dia|Mes| Ano|               Bloco|                Tipo|           Descricao|Descrição_da_localizacao|    Latitude|    Longitude|\n",
      "+---+---+----+--------------------+--------------------+--------------------+------------------------+------------+-------------+\n",
      "| 18|  3|2015|     047XX W OHIO ST|             BATTERY| AGGRAVATED: HANDGUN|                  STREET|41.891398861|-87.744384567|\n",
      "| 18|  3|2015|066XX S MARSHFIEL...|       OTHER OFFENSE|    PAROLE VIOLATION|                  STREET|41.773371528|-87.665319468|\n",
      "| 18|  3|2015|044XX S LAKE PARK...|             BATTERY|DOMESTIC BATTERY ...|               APARTMENT| 41.81386068|-87.596642837|\n",
      "| 18|  3|2015|051XX S MICHIGAN AVE|             BATTERY|              SIMPLE|               APARTMENT|41.800802415|-87.622619343|\n",
      "| 18|  3|2015|    047XX W ADAMS ST|             ROBBERY|      ARMED: HANDGUN|                SIDEWALK|41.878064761|-87.743354013|\n",
      "| 18|  3|2015| 049XX S DREXEL BLVD|             BATTERY|              SIMPLE|               APARTMENT|41.805443345|-87.604283976|\n",
      "| 18|  3|2015|   070XX S MORGAN ST|             BATTERY|DOMESTIC BATTERY ...|               APARTMENT|41.766402779|-87.649296123|\n",
      "| 18|  3|2015| 042XX S PRAIRIE AVE|             BATTERY|DOMESTIC BATTERY ...|               APARTMENT|41.817552577|-87.619818523|\n",
      "| 18|  3|2015| 036XX S WOLCOTT AVE|           NARCOTICS|POSS: CANNABIS 30...|                  STREET|41.828138428|-87.672782106|\n",
      "| 18|  3|2015| 097XX S PRAIRIE AVE|             BATTERY|              SIMPLE|    RESIDENCE PORCH/H...| 41.71745472|-87.617663257|\n",
      "| 18|  3|2015|130XX S DR MARTIN...|     CRIMINAL DAMAGE|          TO VEHICLE|    PARKING LOT/GARAG...|41.658138493|-87.613672862|\n",
      "| 15|  3|2015|078XX S VINCENNES...|       OTHER OFFENSE|HARASSMENT BY TEL...|    CTA GARAGE / OTHE...|41.752406801|-87.633792381|\n",
      "| 18|  3|2015|086XX S EXCHANGE AVE|   WEAPONS VIOLATION|UNLAWFUL POSS OF ...|    DRIVEWAY - RESIDE...|41.738563465|-87.552678593|\n",
      "| 18|  3|2015| 014XX S ASHLAND AVE|             BATTERY|              SIMPLE|                SIDEWALK| 41.86304084|-87.666288555|\n",
      "| 18|  3|2015| 051XX W CHICAGO AVE|               THEFT|        RETAIL THEFT|             GAS STATION|41.894945606|-87.754874977|\n",
      "| 18|  3|2015|077XX S KINGSTON AVE|            BURGLARY|      FORCIBLE ENTRY|               APARTMENT|41.754602618|-87.562650741|\n",
      "| 18|  3|2015|   024XX W NORTH AVE| MOTOR VEHICLE THEFT|          AUTOMOBILE|                   OTHER|41.910312648|-87.687806494|\n",
      "| 18|  3|2015| 069XX S LOOMIS BLVD|               THEFT|       FROM BUILDING|      GROCERY FOOD STORE|41.768167414|-87.659053795|\n",
      "| 18|  3|2015|105XX S LAFAYETTE...|PUBLIC PEACE VIOL...|    RECKLESS CONDUCT|                   ALLEY| 41.70284845|-87.624588931|\n",
      "| 18|  3|2015| 087XX S KIMBARK AVE|               THEFT|       FROM BUILDING|           BAR OR TAVERN|41.736588206| -87.59299436|\n",
      "+---+---+----+--------------------+--------------------+--------------------+------------------------+------------+-------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.sql.functions import col, avg\n",
    "import pyspark.sql.functions as func\n",
    "from pyspark.sql.functions import countDistinct, avg, stddev"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Quantidade de crimes por ano?\n",
    "----\n",
    "\n",
    "* Usando Select para formatar a data da coluna Ano e definindo meu alias para Ano\n",
    "* Em seguida passei o agrupamento por Ano \n",
    "* contei os ocorridos por ano."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+\n",
      "|       Ano|  count|\n",
      "+----------+-------+\n",
      "|01/01/2011| 359474|\n",
      "|01/01/2004| 482572|\n",
      "|01/01/2006| 451114|\n",
      "|01/01/2013| 308550|\n",
      "|01/01/2002| 478166|\n",
      "|01/01/2001| 497358|\n",
      "|01/01/2007| 447292|\n",
      "|      null|6418484|\n",
      "|01/01/2008| 425068|\n",
      "|01/01/2018| 120884|\n",
      "|01/01/2014| 279712|\n",
      "|01/01/2005| 459528|\n",
      "|01/01/2015| 266472|\n",
      "|01/01/2003| 489758|\n",
      "|01/01/2012| 346588|\n",
      "|01/01/2010| 382404|\n",
      "|01/01/2009| 393972|\n",
      "|01/01/2017| 270630|\n",
      "|01/01/2016| 279158|\n",
      "+----------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(F.date_format('Ano','dd/MM/yyyy').alias('Ano')).groupby('Ano').count().show() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Quantidade de crimes por ano que sejam do tipo NARCOTICS?\n",
    "---\n",
    "\n",
    "#### Criando uma View temporaria em memoria com o dataframe "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.createOrReplaceTempView(\"ocorrencias\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Usando SQL para afzer as querys \n",
    "\n",
    "* Primeiro defini meu target(ano)\n",
    "* Depois passei um count(*) para pegar todas as ocorrencias \n",
    "* Como Condicional passei o tipo da ocorrencias NARCOTICS \n",
    "* Agrupando por Ano e Ordenando por Ano"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "CrimesPorAnoNarc = spark.sql(\"SELECT  Ano, COUNT(*) as ocorrencias FROM ocorrencias WHERE Tipo = 'NARCOTICS' GROUP BY Ano ORDER BY Ano\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----------+\n",
      "| Ano|ocorrencias|\n",
      "+----+-----------+\n",
      "|2001|      50996|\n",
      "|2002|      50086|\n",
      "|2003|      55342|\n",
      "|2004|      58500|\n",
      "|2005|      57266|\n",
      "|2006|      55966|\n",
      "|2007|      55936|\n",
      "|2008|      45876|\n",
      "|2009|      42708|\n",
      "|2010|      44344|\n",
      "|2011|      38916|\n",
      "|2012|      36490|\n",
      "|2013|      34778|\n",
      "|2014|      29752|\n",
      "|2015|      21986|\n",
      "|2016|      13988|\n",
      "|2017|      11902|\n",
      "|2018|       5760|\n",
      "+----+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "CrimesPorAnoNarc.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Quantidade de crimes por ano que sejam do tipo NARCOTICS e tenham ocorrido em dias pares?\n",
    "---\n",
    "* Primeiro defini meu target(ano)\n",
    "* Depois passei um count(*) para pegar todas as ocorrencias \n",
    "* Como Condicional passei o tipo da ocorrencias NARCOTICS \n",
    "* Agrupando por Ano e Ordenando por Ano\n",
    "* Passei o boolean type AND com o Parametro Dia % 2 == 0\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = \"\"\"SELECT Ano, COUNT(*) as ocorrencias FROM ocorrencias WHERE Tipo = 'NARCOTICS' AND (Dia % 2) == 0  GROUP BY Ano  ORDER BY Ano \"\"\"                \n",
    "\n",
    "CrimesPorAnoNarcParDay = spark.sql(sql)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----------+\n",
      "| Ano|ocorrencias|\n",
      "+----+-----------+\n",
      "|2001|      24732|\n",
      "|2002|      24610|\n",
      "|2003|      26726|\n",
      "|2004|      28296|\n",
      "|2005|      27610|\n",
      "|2006|      27648|\n",
      "|2007|      27736|\n",
      "|2008|      22302|\n",
      "|2009|      20894|\n",
      "|2010|      21632|\n",
      "|2011|      19330|\n",
      "|2012|      17884|\n",
      "|2013|      16822|\n",
      "|2014|      14644|\n",
      "|2015|      10746|\n",
      "|2016|       6690|\n",
      "|2017|       5732|\n",
      "|2018|       2678|\n",
      "+----+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "CrimesPorAnoNarcParDay.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Mês com maior ocorrência de crimes?\n",
    "----\n",
    "\n",
    "Passei o Parametro Mes dentro da subquery contando o numero de linhas dos Mes e chamando de ocorrencias com isso agrupei por Mes e ordenei para apresentar de ordem descendente, limitando apenas 1 unico valor de saida"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|Mes|\n",
      "+---+\n",
      "|  7|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql = \"select Mes from (select Mes, count(Mes) as ocorrencias from ocorrencias group by Mes order by 'ocorrencias' desc) LIMIT 1\"\n",
    "\n",
    "MesMaiorOcorrencia = spark.sql(sql)\n",
    "MesMaiorOcorrencia.show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Mês com a maior média de ocorrência de crimes?\n",
    "----\n",
    "\n",
    "Aqui fiz 4 querys \n",
    "\n",
    "* 1 - Para agrupar por mes a media score das ocorrencias \n",
    "* 2 - A segunda pega o maximo valor dessa avg_ocorrencias \n",
    "* 3 - Faz um select de mes, AVG(total_linhas) agrupando por Mes \n",
    "* 4 - Ultima query ele pega os Valores Maximos de Mes, e Avg_ocorre, usando condicional where para passar a segunda query com condição para avg_ocorre = MAX valor da média"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "sum_count = df.count() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "13157184\n"
     ]
    }
   ],
   "source": [
    "print(sum_count)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+\n",
      "|Mes| avg_ocorre|\n",
      "+---+-----------+\n",
      "|  7|1.3157184E7|\n",
      "+---+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql = \"SELECT Mes, avg_ocorre  FROM (SELECT Mes, AVG(13157184) AS avg_ocorre FROM ocorrencias GROUP BY Mes)   WHERE avg_ocorre = (SELECT MAX(avg_ocorre) FROM (SELECT Mes, AVG(13157184) AS avg_ocorre FROM ocorrencias GROUP BY Mes)) LIMIT 1\"\n",
    "spark.sql(sql).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Mês por ano com a maior ocorrência de crimes?\n",
    "\n",
    "---\n",
    "\n",
    "* Aqui passei um groubBy(mes) para grupar por mes\n",
    "* Em Seguida passei um agregador agg para o valor Max do Ano\n",
    "* Limitando apenas uma unica saida "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "|Mes|max(Ano)|\n",
      "+---+--------+\n",
      "|  7|    2017|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"Mes\").agg(F.max(\"Ano\")).limit(1).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Mês com a maior ocorrência de crimes do tipo DECEPTIVE PRACTICE?\n",
    "---\n",
    "\n",
    "Aqui pensei que poderia rodar um test checando pelo pyspark e tambem com sql para validar as duas saidas\n",
    "\n",
    "\n",
    "A Primeira DF_DECEPTIVE_PRACTICE passei um filter para pegar apenas os tipos DECEPTIVE PRACTICE \n",
    "\n",
    "Em Seguida agrupei Mes e Ano usando agregação para contar o as vezes que os dados da coluna mes aparece chamando de Mes_ocorr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "DF_DECEPTIVE_PRACTICE = df.filter(df['Tipo']  == \"DECEPTIVE PRACTICE\")\n",
    "DF_DECEPTIVE_PRACTICE = DF_DECEPTIVE_PRACTICE.groupby(['Mes', 'Ano']).agg(F.count(\"Mes\").alias(\"Mes_Ocorr\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Com a DF criado passei os parametros MAX em todas as dimenções de atenção da tabela(Mes, Ano, Mes_Ocorr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(max(Ano)='2018', max(Mes_Ocorr)=1824, max(Mes)='9')]\n"
     ]
    }
   ],
   "source": [
    "value = DF_DECEPTIVE_PRACTICE.agg({\"Mes_Ocorr\": \"max\",\"Mes\":\"max\",\"Ano\":\"Max\"}).collect()\n",
    "print(value)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Validação com SQL\n",
    "\n",
    "Com o dado do dataframe pensei em passar pelo sql para validar a saida dos dados pela query a baixo\n",
    "\n",
    "Passei um MAX(mes, Mes_, Ano) e em uma subquery passei os parametros e Mes,Ano e Count(Mes)\n",
    "\n",
    "Usando WHERE passei o tipo do dado buscado e agrupei por Mes e Ano"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+--------------+\n",
      "|Mes_|Ano_|max(Mes_ocorr)|\n",
      "+----+----+--------------+\n",
      "|   9|2018|          1824|\n",
      "+----+----+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sql = \"\"\"\n",
    "select max(Mes) as Mes_, max(Ano) as Ano_, max(Mes_ocorr) from\n",
    "(select Mes,Ano,count(Mes) as Mes_ocorr FROM ocorrencias WHERE Tipo = 'DECEPTIVE PRACTICE' GROUP BY Mes, Ano)\n",
    "\n",
    "\"\"\"\n",
    "spark.sql(sql).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dia do ano com a maior ocorrência de crimes?\n",
    "\n",
    "----\n",
    "\n",
    "PS: Por falha e erro jo job não consegui rodar essa query\n",
    "\n",
    "Contudo a ideia era agrupar Mes, Ano e Dia e ordenar por ocorrencias e com isso fazer um count do mes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "ename": "Py4JJavaError",
     "evalue": "An error occurred while calling o84.showString.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 15 in stage 40.0 failed 1 times, most recent failure: Lost task 15.0 in stage 40.0 (TID 1718, sandbox-host.hortonworks.com, executor driver): java.io.FileNotFoundException: /tmp/blockmgr-ff9c78ab-fdb4-43b1-846f-2de9dc87e1a6/14/temp_shuffle_4c9dfea7-cb4a-4cad-872b-5bbaf1d38b7d (Too many open files)\n\tat java.io.FileOutputStream.open0(Native Method)\n\tat java.io.FileOutputStream.open(FileOutputStream.java:270)\n\tat java.io.FileOutputStream.<init>(FileOutputStream.java:213)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.initialize(DiskBlockObjectWriter.scala:105)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.open(DiskBlockObjectWriter.scala:118)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.write(DiskBlockObjectWriter.scala:245)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:158)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2059)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2008)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2007)\n\tat scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)\n\tat scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)\n\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2007)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:973)\n\tat scala.Option.foreach(Option.scala:407)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2239)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2188)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2177)\n\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:775)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2099)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2120)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2139)\n\tat org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:467)\n\tat org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:420)\n\tat org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:47)\n\tat org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3627)\n\tat org.apache.spark.sql.Dataset.$anonfun$head$1(Dataset.scala:2697)\n\tat org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3618)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)\n\tat org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)\n\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:764)\n\tat org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)\n\tat org.apache.spark.sql.Dataset.withAction(Dataset.scala:3616)\n\tat org.apache.spark.sql.Dataset.head(Dataset.scala:2697)\n\tat org.apache.spark.sql.Dataset.take(Dataset.scala:2904)\n\tat org.apache.spark.sql.Dataset.getRows(Dataset.scala:300)\n\tat org.apache.spark.sql.Dataset.showString(Dataset.scala:337)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.lang.reflect.Method.invoke(Method.java:498)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n\tat py4j.Gateway.invoke(Gateway.java:282)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: java.io.FileNotFoundException: /tmp/blockmgr-ff9c78ab-fdb4-43b1-846f-2de9dc87e1a6/14/temp_shuffle_4c9dfea7-cb4a-4cad-872b-5bbaf1d38b7d (Too many open files)\n\tat java.io.FileOutputStream.open0(Native Method)\n\tat java.io.FileOutputStream.open(FileOutputStream.java:270)\n\tat java.io.FileOutputStream.<init>(FileOutputStream.java:213)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.initialize(DiskBlockObjectWriter.scala:105)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.open(DiskBlockObjectWriter.scala:118)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.write(DiskBlockObjectWriter.scala:245)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:158)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\t... 1 more\n",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-25-1b7c0c0da4b4>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      8\u001b[0m         ORDER BY 'ocorrencias' DESC) LIMIT 1\n\u001b[1;32m      9\u001b[0m \"\"\"\n\u001b[0;32m---> 10\u001b[0;31m \u001b[0mspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msql\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msql\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     11\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     12\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.7/site-packages/pyspark/sql/dataframe.py\u001b[0m in \u001b[0;36mshow\u001b[0;34m(self, n, truncate, vertical)\u001b[0m\n\u001b[1;32m    438\u001b[0m         \"\"\"\n\u001b[1;32m    439\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtruncate\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mbool\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mand\u001b[0m \u001b[0mtruncate\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 440\u001b[0;31m             \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jdf\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mshowString\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mn\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m20\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mvertical\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    441\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    442\u001b[0m             \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jdf\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mshowString\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mn\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtruncate\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mvertical\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.7/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1303\u001b[0m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1304\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1305\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1306\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1307\u001b[0m         \u001b[0;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.7/site-packages/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    126\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mdeco\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    127\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 128\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    129\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    130\u001b[0m             \u001b[0mconverted\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mconvert_exception\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0me\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjava_exception\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.local/lib/python3.7/site-packages/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[0;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[1;32m    326\u001b[0m                 raise Py4JJavaError(\n\u001b[1;32m    327\u001b[0m                     \u001b[0;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 328\u001b[0;31m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[1;32m    329\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    330\u001b[0m                 raise Py4JError(\n",
      "\u001b[0;31mPy4JJavaError\u001b[0m: An error occurred while calling o84.showString.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 15 in stage 40.0 failed 1 times, most recent failure: Lost task 15.0 in stage 40.0 (TID 1718, sandbox-host.hortonworks.com, executor driver): java.io.FileNotFoundException: /tmp/blockmgr-ff9c78ab-fdb4-43b1-846f-2de9dc87e1a6/14/temp_shuffle_4c9dfea7-cb4a-4cad-872b-5bbaf1d38b7d (Too many open files)\n\tat java.io.FileOutputStream.open0(Native Method)\n\tat java.io.FileOutputStream.open(FileOutputStream.java:270)\n\tat java.io.FileOutputStream.<init>(FileOutputStream.java:213)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.initialize(DiskBlockObjectWriter.scala:105)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.open(DiskBlockObjectWriter.scala:118)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.write(DiskBlockObjectWriter.scala:245)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:158)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2059)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2008)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2007)\n\tat scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)\n\tat scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)\n\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2007)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:973)\n\tat scala.Option.foreach(Option.scala:407)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:973)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2239)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2188)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2177)\n\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:775)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2099)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2120)\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:2139)\n\tat org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:467)\n\tat org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:420)\n\tat org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:47)\n\tat org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3627)\n\tat org.apache.spark.sql.Dataset.$anonfun$head$1(Dataset.scala:2697)\n\tat org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3618)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)\n\tat org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)\n\tat org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)\n\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:764)\n\tat org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)\n\tat org.apache.spark.sql.Dataset.withAction(Dataset.scala:3616)\n\tat org.apache.spark.sql.Dataset.head(Dataset.scala:2697)\n\tat org.apache.spark.sql.Dataset.take(Dataset.scala:2904)\n\tat org.apache.spark.sql.Dataset.getRows(Dataset.scala:300)\n\tat org.apache.spark.sql.Dataset.showString(Dataset.scala:337)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.lang.reflect.Method.invoke(Method.java:498)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n\tat py4j.Gateway.invoke(Gateway.java:282)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: java.io.FileNotFoundException: /tmp/blockmgr-ff9c78ab-fdb4-43b1-846f-2de9dc87e1a6/14/temp_shuffle_4c9dfea7-cb4a-4cad-872b-5bbaf1d38b7d (Too many open files)\n\tat java.io.FileOutputStream.open0(Native Method)\n\tat java.io.FileOutputStream.open(FileOutputStream.java:270)\n\tat java.io.FileOutputStream.<init>(FileOutputStream.java:213)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.initialize(DiskBlockObjectWriter.scala:105)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.open(DiskBlockObjectWriter.scala:118)\n\tat org.apache.spark.storage.DiskBlockObjectWriter.write(DiskBlockObjectWriter.scala:245)\n\tat org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:158)\n\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:127)\n\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)\n\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\t... 1 more\n"
     ]
    }
   ],
   "source": [
    "sql = \"\"\"\n",
    "  SELECT Mes, Ano, Dia  FROM ( \n",
    "        SELECT Mes, \n",
    "        COUNT(Mes) as ocorrencias, \n",
    "        Ano,\n",
    "        Dia  \n",
    "        FROM ocorrencias  GROUP BY Mes, Ano, Dia \n",
    "        ORDER BY 'ocorrencias' DESC) LIMIT 1\n",
    "\"\"\"\n",
    "spark.sql(sql).show()\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}