07_ntb_AnalyticsWithPySpark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyN4SX/T/Not0PzWIIgcvdUa",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/luasampaio/data-engineering/blob/main/7_ntb_AnalyticsWithPySpark.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "source": [
        "!pip install pyspark\n",
        "from pyspark.sql import SparkSession\n",
        "from google.colab import drive\n",
        "\n",
        "# Mount Google Drive\n",
        "drive.mount('/content/drive')\n",
        "\n",
        "# Create a SparkSession\n",
        "spark = SparkSession.builder.appName(\"MySparkApp\").getOrCreate()\n",
        "\n",
        "# Verify the file path after mounting\n",
        "!ls /content/drive/MyDrive/Maga/\n",
        "\n",
        "# Update with the correct path if needed\n",
        "file_path = \"/content/drive/MyDrive/Maga/tab_venda_servicos.csv\"\n",
        "\n",
        "df = (\n",
        "    spark.read.format(\"csv\")\n",
        "    .option(\"header\", True)\n",
        "    .load(file_path)\n",
        ")"
      ],
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "7bqmaDVKlRjp",
        "outputId": "62dcad41-b2b8-4379-b063-b9bead7ea7c8"
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Requirement already satisfied: pyspark in /usr/local/lib/python3.10/dist-packages (3.5.4)\n",
            "Requirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark) (0.10.9.7)\n",
            "Drive already mounted at /content/drive; to attempt to forcibly remount, call drive.mount(\"/content/drive\", force_remount=True).\n",
            "json_transacoes_contas.txt\tresultado_final.csv  servico.csv\t     TESTE_LABS.sql\n",
            "Labs_Maga_LucianaSampaio.ipynb\troubo_furto.csv      tab_venda_servicos.csv\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "df.printSchema()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "D0H1X7PAlqWl",
        "outputId": "58065319-9f8a-4053-f7d1-6344e42c97e3"
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "root\n",
            " |-- data_venda: string (nullable = true)\n",
            " |-- cliente: string (nullable = true)\n",
            " |-- filial_venda: string (nullable = true)\n",
            " |-- servico: string (nullable = true)\n",
            " |-- valor_total: string (nullable = true)\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "df.show(5)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "NxR_OKK_lx0m",
        "outputId": "1598bb63-2071-4883-e079-5aa2e16c5e05"
      },
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+--------+------------+-------+-----------+\n",
            "|data_venda| cliente|filial_venda|servico|valor_total|\n",
            "+----------+--------+------------+-------+-----------+\n",
            "|2022-10-22| 6309189|         256|      5|     232.27|\n",
            "|2023-01-30|90585447|         256|      5|     425.13|\n",
            "|2022-12-09|10733220|         256|      5|     122.67|\n",
            "|2023-05-26| 6338901|         256|      5|     166.68|\n",
            "|2023-06-30| 6278693|         256|      5|     509.95|\n",
            "+----------+--------+------------+-------+-----------+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql.functions import to_date, year, col\n",
        "\n",
        "# Transformar a coluna 'data_venda' para o tipo Date\n",
        "df = df.withColumn(\"data_venda\", to_date(\"data_venda\", \"yyyy-MM-dd\"))\n",
        "\n",
        "# Converter a coluna 'valor_total' para o tipo Integer\n",
        "df = df.withColumn(\"valor_total\", col(\"valor_total\").cast(\"int\"))\n",
        "\n",
        "# Criar a coluna 'ano' extraindo o ano de 'data_venda'\n",
        "df = df.withColumn(\"ano\", year(\"data_venda\"))\n",
        "\n",
        "# Exibir o esquema e as primeiras linhas\n",
        "df.printSchema()\n",
        "df.show(5)\n",
        "\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "C_Vc_lBGqxMx",
        "outputId": "f9ccc778-8719-47d3-e474-68f682d388d1"
      },
      "execution_count": 8,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "root\n",
            " |-- data_venda: date (nullable = true)\n",
            " |-- cliente: string (nullable = true)\n",
            " |-- filial_venda: string (nullable = true)\n",
            " |-- servico: string (nullable = true)\n",
            " |-- valor_total: integer (nullable = true)\n",
            " |-- ano: integer (nullable = true)\n",
            "\n",
            "+----------+--------+------------+-------+-----------+----+\n",
            "|data_venda| cliente|filial_venda|servico|valor_total| ano|\n",
            "+----------+--------+------------+-------+-----------+----+\n",
            "|2022-10-22| 6309189|         256|      5|        232|2022|\n",
            "|2023-01-30|90585447|         256|      5|        425|2023|\n",
            "|2022-12-09|10733220|         256|      5|        122|2022|\n",
            "|2023-05-26| 6338901|         256|      5|        166|2023|\n",
            "|2023-06-30| 6278693|         256|      5|        509|2023|\n",
            "+----------+--------+------------+-------+-----------+----+\n",
            "only showing top 5 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "df.printSchema()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "wQHgpiISqm8t",
        "outputId": "c076b5e8-f954-41b2-9850-2c828d70bdc6"
      },
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "root\n",
            " |-- data_venda: date (nullable = true)\n",
            " |-- cliente: string (nullable = true)\n",
            " |-- filial_venda: string (nullable = true)\n",
            " |-- servico: string (nullable = true)\n",
            " |-- valor_total: integer (nullable = true)\n",
            " |-- ano: integer (nullable = true)\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "#df = df.withColumnRenamed(\"data\", \"data_venda\")\n",
        "#df.printSchema()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "2d_UVXW1rZ-j",
        "outputId": "edb42665-08b0-41cf-d885-15bf05517ccc"
      },
      "execution_count": 6,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "root\n",
            " |-- data_venda: date (nullable = true)\n",
            " |-- cliente: string (nullable = true)\n",
            " |-- filial_venda: string (nullable = true)\n",
            " |-- servico: string (nullable = true)\n",
            " |-- valor_total: string (nullable = true)\n",
            " |-- ano: integer (nullable = true)\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "TotalAno = df.groupBy('ano').count().show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "g2yDkQg-tDk7",
        "outputId": "97891bf6-2d68-46f1-8355-01cc19ae1b0f"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----+-------+\n",
            "| ano|  count|\n",
            "+----+-------+\n",
            "|2023|1090543|\n",
            "|2022|1238715|\n",
            "+----+-------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "TotalAno = df.groupBy('ano').sum('valor_total').show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "HQAQ-yjKuENk",
        "outputId": "50ee5076-ef60-407f-9ddf-b676c43aa6e0"
      },
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----+----------------+\n",
            "| ano|sum(valor_total)|\n",
            "+----+----------------+\n",
            "|2023|       322582388|\n",
            "|2022|       355024591|\n",
            "+----+----------------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "Remover espaços nas colunas tipo string."
      ],
      "metadata": {
        "id": "4dxiF_pulktU"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "print(\"Total registros:\",df.count())"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "7nxRVWdAujjL",
        "outputId": "e783aa67-9047-4e59-db17-63c20e4afad1"
      },
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Total registros: 2329258\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "**Serviços Mais Usados **"
      ],
      "metadata": {
        "id": "JDDFrsK1uz3X"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "ranking = df.groupBy('servico').count().orderBy('count', ascending=False)\n",
        "ranking.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "8s5CLUP8umVl",
        "outputId": "cf27d259-6150-4a5c-9a21-eeddeaa6d6ab"
      },
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-------+\n",
            "|servico|  count|\n",
            "+-------+-------+\n",
            "|      5|1678748|\n",
            "|     19| 330198|\n",
            "|      7| 320312|\n",
            "+-------+-------+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "string_cols = [c for c, t in df.dtypes if t =='string']\n",
        "from pyspark.sql import functions as f\n",
        "\n",
        "for colname in string_cols:\n",
        "\n",
        "    temp1 = df.withColumn(colname, f.trim(f.col(colname)))\n",
        "    temp2 = temp1.withColumn(colname, f.ltrim(f.col(colname)))\n",
        "    df = temp2.withColumn(colname, f.rtrim(f.col(colname)))"
      ],
      "metadata": {
        "id": "sXThBqiYlEdE"
      },
      "execution_count": 7,
      "outputs": []
    }
  ]
}