E_Null_Detection_and_Clean_using_PySpark.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/aekanun2020/2022-PUB_COC-Data-Science-for-Tourism/blob/main/E_Null_Detection_and_Clean_using_PySpark.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
        "!wget -q https://archive.apache.org/dist/spark/spark-2.4.1/spark-2.4.1-bin-hadoop2.7.tgz\n",
        "!tar xf spark-2.4.1-bin-hadoop2.7.tgz\n",
        "!pip install -q findspark\n",
        "\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark-2.4.1-bin-hadoop2.7\"\n",
        "\n",
        "import findspark\n",
        "findspark.init()\n",
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.builder.master(\"local[*]\").getOrCreate()\n",
        "\n",
        "from pyspark.sql import SparkSession\n",
        "\n",
        "spark = SparkSession.builder \\\n",
        "   .appName(\"Neural Network Model\") \\\n",
        "   .config(\"spark.executor.memory\", \"3gb\") \\\n",
        "   .getOrCreate()\n",
        "   \n",
        "sc = spark.sparkContext\n",
        "\n",
        "sc"
      ],
      "metadata": {
        "id": "Xfx-_ouXlQyi",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 193
        },
        "outputId": "e04b45c2-b775-49e6-ff4b-09a619844c61"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "<SparkContext master=local[*] appName=pyspark-shell>"
            ],
            "text/html": [
              "\n",
              "        <div>\n",
              "            <p><b>SparkContext</b></p>\n",
              "\n",
              "            <p><a href=\"http://25ccfebeaede:4040\">Spark UI</a></p>\n",
              "\n",
              "            <dl>\n",
              "              <dt>Version</dt>\n",
              "                <dd><code>v2.4.1</code></dd>\n",
              "              <dt>Master</dt>\n",
              "                <dd><code>local[*]</code></dd>\n",
              "              <dt>AppName</dt>\n",
              "                <dd><code>pyspark-shell</code></dd>\n",
              "            </dl>\n",
              "        </div>\n",
              "        "
            ]
          },
          "metadata": {},
          "execution_count": 5
        }
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "7oO8LTWYlGg7"
      },
      "outputs": [],
      "source": [
        "from pyspark.sql import functions as sparkf\n",
        "\n",
        "raw_df = spark.createDataFrame(\n",
        " [\n",
        "  ('Store 1',1,448),\n",
        "  ('Store 1',2,None),\n",
        "  ('Store 1',3,499),\n",
        "  ('Store 1',44,432),\n",
        "  (None,None,None),\n",
        "  ('Store 2',1,355),\n",
        "  ('Store 2',1,355),\n",
        "  ('Store 2',None,345),\n",
        "  ('Store 2',3,387),\n",
        "  ('Store 2',4,312)\n",
        "],\n",
        " ['Store','WeekInMonth','Revenue']\n",
        ")\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "b1sLmRZalGhC",
        "outputId": "63058045-e272-4fc8-9aee-91d3edcc17e5",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "root\n",
            " |-- Store: string (nullable = true)\n",
            " |-- WeekInMonth: long (nullable = true)\n",
            " |-- Revenue: long (nullable = true)\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.printSchema()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "yViuD4PJlGhD",
        "outputId": "bb2edb3b-4a3c-469a-fef3-7ca1d283a23c",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "10"
            ]
          },
          "metadata": {},
          "execution_count": 8
        }
      ],
      "source": [
        "raw_df.count()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "BJFKWNYtlGhE",
        "outputId": "81db85f7-251f-4508-f58b-0ac75d9b5a15",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          2|   null|\n",
            "|Store 1|          3|    499|\n",
            "|Store 1|         44|    432|\n",
            "|   null|       null|   null|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|       null|    345|\n",
            "|Store 2|          3|    387|\n",
            "|Store 2|          4|    312|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "wo9TXhs6lGhF",
        "outputId": "7d85d305-f3b3-4ede-decf-5297038bd275",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-------+------------------+------------------+\n",
            "|summary|  Store|       WeekInMonth|           Revenue|\n",
            "+-------+-------+------------------+------------------+\n",
            "|  count|      9|                 8|                 8|\n",
            "|   mean|   null|             7.375|           391.625|\n",
            "| stddev|   null|14.841423689890979|62.741960213469355|\n",
            "|    min|Store 1|                 1|               312|\n",
            "|    max|Store 2|                44|               499|\n",
            "+-------+-------+------------------+------------------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.describe().show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "joP71uKHlGhG",
        "outputId": "e7103fb7-95b8-4be1-bff8-b40868202b13",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----+-----------+-------+\n",
            "|Store|WeekInMonth|Revenue|\n",
            "+-----+-----------+-------+\n",
            "|    1|          2|      2|\n",
            "+-----+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.select(\n",
        "  [sparkf.count(sparkf.when(sparkf.isnull(c), c)).alias(c) for c in raw_df.columns]\n",
        ").show()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5VjMT_cylGhH",
        "outputId": "ba51bd03-a386-4283-a118-cf4dbfbc853c",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          2|   null|\n",
            "|   null|       null|   null|\n",
            "|Store 2|       null|    345|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "#ระบุว่า Nulls อยู่ที่ row ไหน - เพื่อดูว่าเป็น MCAR / MAR / MNAR / Structural Missing\n",
        "\n",
        "from functools import reduce\n",
        "\n",
        "raw_df.filter(reduce(lambda a1, a2: a1 | a2,\\\n",
        "                     (sparkf.col(c).isNull() \\\n",
        "                      for c in raw_df.columns))).show()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "-SihbWuMlGhJ",
        "outputId": "3ced0170-acd6-4e93-96b1-337a34b5b22a",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          2|   null|\n",
            "|Store 1|          3|    499|\n",
            "|Store 1|         44|    432|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|       null|    345|\n",
            "|Store 2|          3|    387|\n",
            "|Store 2|          4|    312|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "#ซ่อน/ลบ เฉพาะ row ที่มีค่า nulls ในทุก attributes แล้วซ่อนค่า nulls จากนั้นส่งให้กับตัวแปรใหม่ [III]\n",
        "\n",
        "noNullRow_Df = raw_df.dropna('all')\n",
        "\n",
        "noNullRow_Df.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "LsoHplhxlGhK",
        "outputId": "7836de23-7e17-478d-f453-f680e0b954b9",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          2|   null|\n",
            "|Store 1|          3|    499|\n",
            "|Store 1|         44|    432|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          3|    387|\n",
            "|Store 2|          4|    312|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "#ซ่อน/ลบ row ที่มีค่า nulls ใน attributes ใด attribute หนึ่ง (หรือทุก attributes) จาก subset [IV]\n",
        "\n",
        "\n",
        "raw_df.dropna(how='any', \\\n",
        "              subset=['Store','WeekInMonth']).show()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "S7Zoua2mlGhL",
        "outputId": "91b6d1f0-1d3f-468f-fe4d-48932804353c",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          3|    499|\n",
            "|Store 1|         44|    432|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          3|    387|\n",
            "|Store 2|          4|    312|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "#ซ่อน/ลบ row ที่มีค่า nulls ใน attributes ใด attribute หนึ่ง (หรือทุก attributes) แล้วส่งให้กับตัวแปรใหม่ [V]\n",
        "\n",
        "raw_df.dropna(how='any').show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "tYMbEn0ilGhL",
        "outputId": "c935a6f4-918a-4b64-872f-5de721e7d6d0",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          2|   9999|\n",
            "|Store 1|          3|    499|\n",
            "|Store 1|         44|    432|\n",
            "|   null|       null|   9999|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|       null|    345|\n",
            "|Store 2|          3|    387|\n",
            "|Store 2|          4|    312|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "#แทนที่ค่า nulls ด้วยเลข 9999 เฉพาะใน “Revenue” [VI]\n",
        "\n",
        "raw_df.fillna(9999,['Revenue']).show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "IKQfhhPjlGhM",
        "outputId": "7ea88949-06f1-484e-881d-f3eb864a777e",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------+-----------+-------+\n",
            "|         Store|WeekInMonth|Revenue|\n",
            "+--------------+-----------+-------+\n",
            "|       Store 1|          1|    448|\n",
            "|       Store 1|          2|      3|\n",
            "|       Store 1|          3|    499|\n",
            "|       Store 1|         44|    432|\n",
            "|Assume_Store 1|          2|      3|\n",
            "|       Store 2|          1|    355|\n",
            "|       Store 2|          1|    355|\n",
            "|       Store 2|          2|    345|\n",
            "|       Store 2|          3|    387|\n",
            "|       Store 2|          4|    312|\n",
            "+--------------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "#แทนที่ค่า nulls ด้วยเลข 0 ในทุก Attributes ซึ่งเป็น numeric และด้วยค่าอื่นๆ สำหรับ string [VII]\n",
        "raw_df.fillna({'Store':'Assume_Store 1',\\\n",
        "              'WeekInMonth':'2','Revenue':3}).show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vOVI8MWNlGhN"
      },
      "outputs": [],
      "source": [
        "# [VIII]\n",
        "from pyspark.sql import functions as sparkf"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "go374A45lGhO",
        "outputId": "9fd0a715-63a5-4baa-f43c-83b66fdfe779",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          2|   null|\n",
            "|Store 1|          3|    499|\n",
            "|Store 1|         44|    432|\n",
            "|   null|       null|   null|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|       null|    345|\n",
            "|Store 2|          3|    387|\n",
            "|Store 2|          4|    312|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "f3UR9MBolGhO",
        "outputId": "d6fb9617-3be2-4acb-cf5b-b60ad8eceb8b",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "root\n",
            " |-- Store: string (nullable = true)\n",
            " |-- WeekInMonth: long (nullable = true)\n",
            " |-- Revenue: long (nullable = true)\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.printSchema()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "7eP_Q9PElGhP"
      },
      "outputs": [],
      "source": [
        "magic_percentile = sparkf.expr('percentile_approx(Revenue, 0.5)')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XANu8GDNlGhQ",
        "outputId": "857921b7-1a4d-4f0e-f8b7-400e9e70a44c",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 2|          4|    312|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          3|    387|\n",
            "|Store 1|         44|    432|\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          3|    499|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.na.drop().orderBy('Revenue').show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ZhOXGHnzlGhR",
        "outputId": "8e882905-42eb-4ea0-c45e-a583719a0c5c",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-------+\n",
            "|  Store|med_val|\n",
            "+-------+-------+\n",
            "|Store 1|    448|\n",
            "+-------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.na.drop().groupBy('Store').agg(magic_percentile.alias('med_val'))\\\n",
        ".filter(sparkf.col('Store') == 'Store 1').show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8nrSd39_lGhR"
      },
      "outputs": [],
      "source": [
        "imputed_value = raw_df.na.drop().groupBy('Store').agg(magic_percentile.alias('med_val'))\\\n",
        ".filter(sparkf.col('Store') == 'Store 1').collect()[0][1]"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "W0IrEkQ9lGhS",
        "outputId": "a96f74e0-c288-4244-ffb9-f0ac87e882b3",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "448"
            ]
          },
          "metadata": {},
          "execution_count": 25
        }
      ],
      "source": [
        "imputed_value"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "KJ3ckfw0lGhS",
        "outputId": "ca648e23-4bb9-4707-fac5-ae4c43d5beaf",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+--------------+\n",
            "|  Store|WeekInMonth|Revenue|noNull_Revenue|\n",
            "+-------+-----------+-------+--------------+\n",
            "|Store 1|          1|    448|           448|\n",
            "|Store 1|          2|   null|           448|\n",
            "|Store 1|          3|    499|           499|\n",
            "|Store 1|         44|    432|           432|\n",
            "|   null|       null|   null|          null|\n",
            "|Store 2|          1|    355|           355|\n",
            "|Store 2|          1|    355|           355|\n",
            "|Store 2|       null|    345|           345|\n",
            "|Store 2|          3|    387|           387|\n",
            "|Store 2|          4|    312|           312|\n",
            "+-------+-----------+-------+--------------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.withColumn('noNull_Revenue'\\\n",
        "                  ,sparkf.when((sparkf.col('Store')=='Store 1')\\\n",
        "                &(sparkf.col('Revenue').isNull()),imputed_value).otherwise(sparkf.col('Revenue'))).show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6ryAhrFtlGhT",
        "outputId": "4afa3d8a-b39e-425d-b386-4b764555b96e",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-------+-----------+-------+\n",
            "|  Store|WeekInMonth|Revenue|\n",
            "+-------+-----------+-------+\n",
            "|Store 1|          1|    448|\n",
            "|Store 1|          2|   null|\n",
            "|Store 1|          3|    499|\n",
            "|Store 1|         44|    432|\n",
            "|   null|       null|   null|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|          1|    355|\n",
            "|Store 2|       null|    345|\n",
            "|Store 2|          3|    387|\n",
            "|Store 2|          4|    312|\n",
            "+-------+-----------+-------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "raw_df.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Hf3C_xY0lGhU"
      },
      "outputs": [],
      "source": []
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    },
    "colab": {
      "provenance": [],
      "collapsed_sections": [],
      "include_colab_link": true
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}