ETL_training_2023.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "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/msdevanms/pyspark_training/blob/main/ETL_training_2023.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Initial Setup - Spark Installation"
      ],
      "metadata": {
        "id": "fWKUreoV-UwR"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install pyspark"
      ],
      "metadata": {
        "id": "waElCxaVYZS1"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "oNOtAOvaOLuP"
      },
      "outputs": [],
      "source": [
        "# Create SparkSession and sparkcontext\n",
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.builder\\\n",
        "                    .master(\"local\")\\\n",
        "                    .appName('Firstprogram')\\\n",
        "                    .getOrCreate()\n",
        "sc=spark.sparkContext"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Load CSV data"
      ],
      "metadata": {
        "id": "E6Yt0GbQ-iGa"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "! wget 'https://raw.githubusercontent.com/msdevanms/pyspark_training/main/data/data_descriptive.csv'"
      ],
      "metadata": {
        "id": "vAGEuIbGJphY"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "mydataframe = spark.read.format(\"csv\").option(\"header\",\"true\").load(\"data_descriptive.csv\")"
      ],
      "metadata": {
        "id": "Mg-N_adSQzly"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "mydataframe.show()"
      ],
      "metadata": {
        "id": "B0fOi1zlRWqc"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Inspect Data"
      ],
      "metadata": {
        "id": "jATsjs-O-pKy"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "mydataframe.count()"
      ],
      "metadata": {
        "id": "VUOUMCVs_s-J"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "mydataframe.printSchema()"
      ],
      "metadata": {
        "id": "ecmSvTsp_xGJ"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "mydataframe.describe().show()"
      ],
      "metadata": {
        "id": "SyXNMFr1Rkmk"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Setting Schema"
      ],
      "metadata": {
        "id": "Ra9LTImPCLo6"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Setting schema and importing:\n",
        "from pyspark.sql.types import *\n",
        "myschema = StructType([\n",
        "      StructField('id', IntegerType()),\n",
        "      StructField('first_name', StringType()),\n",
        "      StructField('last_name', StringType()),\n",
        "      StructField('gender', StringType()),\n",
        "      StructField('City', StringType()),\n",
        "      StructField('JobTitle', StringType()),\n",
        "      StructField('Salary', StringType()),\n",
        "      StructField('Latitude', StringType()),\n",
        "      StructField('Longitude', FloatType())\n",
        "])\n",
        "df = spark.read.csv(\"data_descriptive.csv\", header=True, schema = myschema)\n",
        "df.show()"
      ],
      "metadata": {
        "id": "Qebz90vNByIB"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df.describe().show()"
      ],
      "metadata": {
        "id": "2oZgebX1CUsx"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Data Cleansing - Null and Duplicate removal"
      ],
      "metadata": {
        "id": "_OApaHC8-xnR"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "no_duplicate_df = df.dropDuplicates()"
      ],
      "metadata": {
        "id": "Tta8IO-59Qsk"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "no_duplicate_df.count()"
      ],
      "metadata": {
        "id": "qUzcdM9SBKbT"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "no_duplicate_df.sort(no_duplicate_df.id).show()"
      ],
      "metadata": {
        "id": "mJXGMpB79Xc7"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Replace null for 'Unknown':\n",
        "from pyspark.sql.functions import *\n",
        "no_null_df = no_duplicate_df.withColumn(\"clean_city\",when(no_duplicate_df.City.isNull(),'Unknown').otherwise(no_duplicate_df.City))"
      ],
      "metadata": {
        "id": "MtcbNDQj8FYD"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "no_null_df.show(truncate=False)"
      ],
      "metadata": {
        "id": "o9glvQCAY-Qo"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "no_null_df2 = no_null_df.filter(no_null_df.JobTitle.isNotNull())"
      ],
      "metadata": {
        "id": "i56kKK0haSEr"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "no_null_df3 = no_null_df.filter(no_null_df.JobTitle.isNull())"
      ],
      "metadata": {
        "id": "Zy8iD-CYD4hR"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "no_null_df3.show()"
      ],
      "metadata": {
        "id": "IetE2GncD-DR"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "no_null_df2.count()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "MPEckJxXDvuL",
        "outputId": "052a0cb9-36c8-4edd-c582-0f49730fa3e5"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "998"
            ]
          },
          "metadata": {},
          "execution_count": 102
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "no_null_df2.show(truncate=False)"
      ],
      "metadata": {
        "id": "5yMryfWh6anT"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "with_column_df = no_null_df2.withColumn('clean_salary', no_null_df2.Salary.substr(2,100).cast('float'))"
      ],
      "metadata": {
        "id": "er2qDsVOaTHL"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# take(1)[0][0]: take 1st element of df row, column = [0,0]\n",
        "mean = with_column_df.groupBy().avg('clean_salary').take(1)[0][0]"
      ],
      "metadata": {
        "id": "C-jiaGR2aTiL"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "mean"
      ],
      "metadata": {
        "id": "D-c_tekfc4Bx"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "with_column_df.describe().select(\"summary\",\"Salary\", \"clean_salary\").show()"
      ],
      "metadata": {
        "id": "_PAVteJrEUaY"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql.functions import lit\n",
        "with_column_df2 = with_column_df.withColumn('new_salary', when(with_column_df.clean_salary.isNull(), lit(mean)).otherwise(with_column_df.clean_salary))"
      ],
      "metadata": {
        "id": "XZlXMDy8e-rW"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "with_column_df2.show()"
      ],
      "metadata": {
        "id": "ETipd_dSglbV"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "latitudes = with_column_df2.select(\"latitude\")"
      ],
      "metadata": {
        "id": "F2w-4fYJgvG7"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "latitudes.show()"
      ],
      "metadata": {
        "id": "zobte73yvKhi"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "latitudes = latitudes.filter(latitudes.latitude.isNotNull())"
      ],
      "metadata": {
        "id": "rwml4PLQvXaZ"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "latitudes.show()"
      ],
      "metadata": {
        "id": "5P3hAqUaKQoX"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# convert to float:\n",
        "latitudes = latitudes.withColumn('latitude2', latitudes.latitude.cast('float')).select('latitude2')"
      ],
      "metadata": {
        "id": "G-2n2hF8KgtI"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "latitudes.show()"
      ],
      "metadata": {
        "id": "oEuyHyUnKzdU"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import numpy as np\n",
        "median = np.median(latitudes.collect())"
      ],
      "metadata": {
        "id": "njvbNOFIK8ft"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "print(median)"
      ],
      "metadata": {
        "id": "0AG8DWR6LDtQ"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# replace null with median:\n",
        "median_replaced_df = with_column_df2.withColumn('lat', when (with_column_df2.Latitude.isNull(), lit(median)).otherwise(with_column_df2.Latitude))"
      ],
      "metadata": {
        "id": "F66bMDPDLILi"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "median_replaced_df.show()"
      ],
      "metadata": {
        "id": "4iARWJ80L2C2"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Aggregation and Group By"
      ],
      "metadata": {
        "id": "8Qc_gQISL7KB"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import pyspark.sql.functions as sqlfunc\n",
        "genders = median_replaced_df.groupBy('gender').agg(sqlfunc.avg('new_salary').alias('AvgSalary'))"
      ],
      "metadata": {
        "id": "xAlhtdmNL1o1"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Avg salary per gender:\n",
        "genders.show()"
      ],
      "metadata": {
        "id": "snhnSfpZPAJV"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df = median_replaced_df.withColumn('female_salary', when(median_replaced_df.gender=='Female', median_replaced_df.new_salary).otherwise(lit(0)))"
      ],
      "metadata": {
        "id": "4M_-SI0pPVQN"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df.show()"
      ],
      "metadata": {
        "id": "l-HdSOinPnYN"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#lit = literal\n",
        "df = df.withColumn('male_salary', when(df.gender=='Male', df.new_salary).otherwise(lit(0)))"
      ],
      "metadata": {
        "id": "sMZmoRsMPthj"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df.show()"
      ],
      "metadata": {
        "id": "FvBCy0xbQPzT"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Calculate avg salary per jog title:\n",
        "df = df.groupBy('JobTitle').agg(sqlfunc.avg('female_salary').alias('female_salary'),sqlfunc.avg('male_salary').alias('male_salary'))"
      ],
      "metadata": {
        "id": "FUXBn_nZQWn3"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df.show()"
      ],
      "metadata": {
        "id": "5V7wkQFOQuo0"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# difference between gender salary:\n",
        "df = df.withColumn('delta', df.female_salary - df.male_salary)"
      ],
      "metadata": {
        "id": "HJ05G0MJK4BI"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "cityavg = median_replaced_df.groupBy('City').agg(sqlfunc.avg('new_salary').alias('avgsalary'))"
      ],
      "metadata": {
        "id": "WXElq0xALPHC"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "cityavg = cityavg.sort(col('avgsalary').desc())"
      ],
      "metadata": {
        "id": "TS30_QjiNLWC"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Top cities with higher avg salary:\n",
        "cityavg.show()"
      ],
      "metadata": {
        "id": "8TMyoWOkNUCi"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df.write.csv('out_dataframe.csv')\n",
        "df.write.json('out_dataframe.json')\n",
        "df.write.parquet('out_dataframe.parquet')"
      ],
      "metadata": {
        "id": "z4aL0mlSLGCj"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}