SM Desafio.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  BIGDATA -  Ingestão de Dados\n",
    "\n",
    "###  ETL RDD\n",
    "\n",
    "### Claudio Gervasio de Lisboa  ABR/2019"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as f\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql.types import StructField, StructType, StringType, DateType, IntegerType,LongType, DecimalType, DoubleType\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql import SparkSession, SQLContext\n",
    "\n",
    "from pyspark.sql.functions import date_format, dayofmonth, hour, col, max as max_, split, regexp_extract\n",
    "\n",
    "from datetime import datetime, date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Iniciar spark session\n",
    "_spark = SparkSession.builder \\\n",
    "                    .master(\"local\") \\\n",
    "                    .appName(\"ETL\") \\\n",
    "                    .getOrCreate()\n",
    "_sqlContext = SQLContext(_spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Files localization ON Windows\n",
    "_diretorio = 'C:\\\\dump\\\\sm\\\\\\etl_pyspark\\\\dados\\\\mock\\\\nasa'\n",
    "\n",
    "# Diretório Docker teste !!!!\n",
    "_diretorio = '/home/jovyan/work/etl_pyspark/dados/mock/nasa'\n",
    "\n",
    "df_nasa_log  = _sqlContext.read.text(_diretorio)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### RDD Ingestão APACHE LOG"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dataframes\n",
    "df_nasa_view = df_nasa_log.select(regexp_extract('value', r'^([^\\s]+\\s)', 1).alias('host'),\n",
    "                                           regexp_extract('value', r'^.*\\[(\\d\\d/\\w{3}/\\d{4}:\\d{2}:\\d{2}:\\d{2} -\\d{4})]',\n",
    "                                                          1).alias('timestamp'),\n",
    "                                           regexp_extract('value', r'^.*\"\\w+\\s+([^\\s]+)\\s+HTTP.*\"', 1).alias('url'),\n",
    "                                           regexp_extract('value', r'^.*\"\\s+([^\\s]+)', 1).cast('integer').alias(\n",
    "                                               'status'),\n",
    "                                           regexp_extract('value', r'^.*\\s+(\\d+)$', 1).cast('integer').alias(\n",
    "                                               'bytes'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+--------------------------+-----------------------------------------------+------+-----+\n",
      "|host              |timestamp                 |url                                            |status|bytes|\n",
      "+------------------+--------------------------+-----------------------------------------------+------+-----+\n",
      "|in24.inetnebr.com |01/Aug/1995:00:00:01 -0400|/shuttle/missions/sts-68/news/sts-68-mcc-05.txt|200   |1839 |\n",
      "|uplherc.upl.com   |01/Aug/1995:00:00:07 -0400|/                                              |304   |0    |\n",
      "|uplherc.upl.com   |01/Aug/1995:00:00:08 -0400|/images/ksclogo-medium.gif                     |304   |0    |\n",
      "|uplherc.upl.com   |01/Aug/1995:00:00:08 -0400|/images/MOSAIC-logosmall.gif                   |304   |0    |\n",
      "|uplherc.upl.com   |01/Aug/1995:00:00:08 -0400|/images/USA-logosmall.gif                      |304   |0    |\n",
      "+------------------+--------------------------+-----------------------------------------------+------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Tratamento dos dados\n",
    "df_nasa_view.show(n=5,truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### RDD em CACHE para otimizar a leitura das informações em memória"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[host: string, timestamp: string, url: string, status: int, bytes: int]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# RDD Cache\n",
    "\n",
    "df_nasa_view.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data VIEW - 1 - Número de hosts únicos."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Data VIEW - 1 - Número​ ​ de​ ​ hosts​ ​ únicos.\n",
    "def view_d1():\n",
    "    df = df_nasa_view.groupBy('host') \\\n",
    "        .agg(f.count('host').alias('total')) \\\n",
    "        .filter(col('total') == 1) \\\n",
    "        .count()\n",
    "\n",
    "    print('Data VIEW - 1 - Número​ ​ de​ ​ hosts​ ​ únicos = {0}'.format(df))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data VIEW - 1 - Número​ ​ de​ ​ hosts​ ​ únicos = 9269\n"
     ]
    }
   ],
   "source": [
    "view_d1()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Data VIEW - 2. O total de erros 404."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Data VIEW - 2. O total de erros 404.\n",
    "def view_d2():\n",
    "    print('Data VIEW - 2. O total de erros 404.')\n",
    "    df_nasa_view.groupBy('status') \\\n",
    "                .agg(f.count('status').alias('total')) \\\n",
    "                .filter(col('status') == '404') \\\n",
    "                .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data VIEW - 2. O total de erros 404.\n",
      "+------+-----+\n",
      "|status|total|\n",
      "+------+-----+\n",
      "|404   |20901|\n",
      "+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "view_d2()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data View - 3. Os 5 URLs que mais causaram erro 404."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Data View - 3. Os 5 URLs que mais causaram erro 404.\n",
    "def view_d3():\n",
    "    print('Data View - 3. Os 5 URLs que mais causaram erro 404.')\n",
    "    df_nasa_view.groupBy('url','status') \\\n",
    "                .agg(f.count('url').alias('total')) \\\n",
    "                .where( col('status') == '404') \\\n",
    "                .sort(col('total').desc()) \\\n",
    "                .show(n=5, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data View - 3. Os 5 URLs que mais causaram erro 404.\n",
      "+--------------------------------------------+------+-----+\n",
      "|url                                         |status|total|\n",
      "+--------------------------------------------+------+-----+\n",
      "|/pub/winvn/readme.txt                       |404   |2004 |\n",
      "|/pub/winvn/release.txt                      |404   |1732 |\n",
      "|/shuttle/missions/STS-69/mission-STS-69.html|404   |682  |\n",
      "|/shuttle/missions/sts-68/ksc-upclose.gif    |404   |426  |\n",
      "|/history/apollo/a-001/a-001-patch-small.gif |404   |384  |\n",
      "+--------------------------------------------+------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "view_d3()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data View - 4. Quantidade de erros 404 por dia."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Data View - 4. Quantidade de erros 404 por dia.\n",
    "def view_d4():\n",
    "    print('Data View - 4. Quantidade de erros 404 por dia.')\n",
    "    df_nasa_view.withColumn('dia', f.col('timestamp').substr(1,2)) \\\n",
    "                .groupBy('status', col('dia')) \\\n",
    "                .agg(f.count('status').alias('total')) \\\n",
    "                .where( col('status') == '404') \\\n",
    "                .distinct() \\\n",
    "                .sort(col('dia').asc(), col('total').desc()) \\\n",
    "                .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data View - 4. Quantidade de erros 404 por dia.\n",
      "+------+---+-----+\n",
      "|status|dia|total|\n",
      "+------+---+-----+\n",
      "|404   |01 |559  |\n",
      "|404   |02 |291  |\n",
      "|404   |03 |778  |\n",
      "|404   |04 |705  |\n",
      "|404   |05 |733  |\n",
      "|404   |06 |1013 |\n",
      "|404   |07 |1107 |\n",
      "|404   |08 |693  |\n",
      "|404   |09 |627  |\n",
      "|404   |10 |713  |\n",
      "|404   |11 |734  |\n",
      "|404   |12 |667  |\n",
      "|404   |13 |748  |\n",
      "|404   |14 |700  |\n",
      "|404   |15 |581  |\n",
      "|404   |16 |516  |\n",
      "|404   |17 |677  |\n",
      "|404   |18 |721  |\n",
      "|404   |19 |848  |\n",
      "|404   |20 |740  |\n",
      "+------+---+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "view_d4()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data View - 5. O total de bytes retornados."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Data View - 5. O total de bytes retornados.\n",
    "def view_d5():\n",
    "    print('Data View - 5. O​ ​total de​ ​bytes​ retornados.')\n",
    "    df_nasa_view.agg(f.sum('bytes').alias('total_bytes')) \\\n",
    "                .sort(col('total_bytes').desc()) \\\n",
    "                .show(n=5, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data View - 5. O​ ​total de​ ​bytes​ retornados.\n",
      "+-----------+\n",
      "|total_bytes|\n",
      "+-----------+\n",
      "|65524314915|\n",
      "+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "view_d5()"
   ]
  },
  {
   "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.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}