FinalProject.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Analysis using Spark\n",
    "\n",
    "In this project, we focus on mastering Spark SQL. We create a DataFrame from a CSV file and apply transformations and actions using Spark SQL.\n",
    "\n",
    "- Task 1: Generate DataFrame from CSV data.\n",
    "- Task 2: Define a schema for the data.\n",
    "- Task 3: Display schema of DataFrame.\n",
    "- Task 4: Create a temporary view.\n",
    "- Task 5: Execute an SQL query.\n",
    "- Task 6: Calculate Average Salary by Department.\n",
    "- Task 7: Filter and Display IT Department Employees.\n",
    "- Task 8: Add 10% Bonus to Salaries.\n",
    "- Task 9: Find Maximum Salary by Age.\n",
    "- Task 10: Self-Join on Employee Data.\n",
    "- Task 11: Calculate Average Employee Age.\n",
    "- Task 12: Calculate Total Salary by Department.\n",
    "- Task 13: Sort Data by Age and Salary.\n",
    "- Task 14: Count Employees in Each Department.\n",
    "- Task 15: Filter Employees with the letter o in the Name.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Preliminaries: Installing libraries and downloading data\n",
    "\n",
    "Install the required libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting pyspark\n",
      "  Downloading pyspark-3.4.3.tar.gz (311.4 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m311.4/311.4 MB\u001b[0m \u001b[31m1.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m00:01\u001b[0m\n",
      "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25ldone\n",
      "\u001b[?25hCollecting py4j==0.10.9.7 (from pyspark)\n",
      "  Downloading py4j-0.10.9.7-py2.py3-none-any.whl (200 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m200.5/200.5 kB\u001b[0m \u001b[31m32.7 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
      "  Building wheel for pyspark (setup.py) ... \u001b[?25ldone\n",
      "\u001b[?25h  Created wheel for pyspark: filename=pyspark-3.4.3-py2.py3-none-any.whl size=311885504 sha256=c317162eae69c6ff7c94e93ce21715677642093a507b4e62b99303c4f978a8c3\n",
      "  Stored in directory: /home/jupyterlab/.cache/pip/wheels/37/bc/bb/77785f6fcd2c83e663647f73225b76f3a3d5fd00762d7daf6f\n",
      "Successfully built pyspark\n",
      "Installing collected packages: py4j, pyspark\n",
      "Successfully installed py4j-0.10.9.7 pyspark-3.4.3\n",
      "Collecting findspark\n",
      "  Downloading findspark-2.0.1-py2.py3-none-any.whl (4.4 kB)\n",
      "Installing collected packages: findspark\n",
      "Successfully installed findspark-2.0.1\n"
     ]
    }
   ],
   "source": [
    "!pip install pyspark\n",
    "!pip install findspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Download the required data file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'employees.csv'"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Download the CSV data first into a local `employees.csv` file\n",
    "import wget\n",
    "wget.download(\"https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/IBM-BD0225EN-SkillsNetwork/data/employees.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Importing Libraries\n",
    "\n",
    "Importing the required libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import findspark\n",
    "import warnings\n",
    "\n",
    "\n",
    "def warn(*args, **kwargs):\n",
    "    pass\n",
    "\n",
    "\n",
    "# Suppress generated warnings\n",
    "warnings.warn = warn\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "findspark.init()\n",
    "\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a spark session\n",
    "\n",
    "Ignore any warnings by SparkSession command"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/05/02 03:51:50 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Data Analysis using Spark\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tasks\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 1: Generate a Spark DataFrame from the CSV data\n",
    "\n",
    "Our initial task involves reading the downloaded CSV file named `employees.csv` into a Spark DataFrame called `employees_df`.\n",
    "\n",
    "* The `spark.read.csv` function loads the data into a dataframe\n",
    "* The `header=True` indicates that there is a header row in our csv file\n",
    "* The `inferSchema=True` tells spark to automatically determine the data types of the columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+------+---+----------+\n",
      "|Emp_No| Emp_Name|Salary|Age|Department|\n",
      "+------+---------+------+---+----------+\n",
      "|   198|   Donald|  2600| 29|        IT|\n",
      "|   199|  Douglas|  2600| 34|     Sales|\n",
      "|   200| Jennifer|  4400| 36| Marketing|\n",
      "|   201|  Michael| 13000| 32|        IT|\n",
      "|   202|      Pat|  6000| 39|        HR|\n",
      "|   203|    Susan|  6500| 36| Marketing|\n",
      "|   204|  Hermann| 10000| 29|   Finance|\n",
      "|   205|  Shelley| 12008| 33|   Finance|\n",
      "|   206|  William|  8300| 37|        IT|\n",
      "|   100|   Steven| 24000| 39|        IT|\n",
      "|   101|    Neena| 17000| 27|     Sales|\n",
      "|   102|      Lex| 17000| 37| Marketing|\n",
      "|   103|Alexander|  9000| 39| Marketing|\n",
      "|   104|    Bruce|  6000| 38|        IT|\n",
      "|   105|    David|  4800| 39|        IT|\n",
      "|   106|    Valli|  4800| 38|     Sales|\n",
      "|   107|    Diana|  4200| 35|     Sales|\n",
      "|   108|    Nancy| 12008| 28|     Sales|\n",
      "|   109|   Daniel|  9000| 35|        HR|\n",
      "|   110|     John|  8200| 31| Marketing|\n",
      "+------+---------+------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Read data from the \"emp\" CSV file and import it into a DataFrame variable named \"employees_df\"\n",
    "employees_df = spark.read.csv(\"employees.csv\", header=True, inferSchema=True)\n",
    "\n",
    "# Display the dataframe content\n",
    "employees_df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 2: Define a schema for the data\n",
    "\n",
    "Next, we define a schema for the input data and then use the user-defined schema to read the CSV file named `employees.csv` into a Spark DataFrame called `employees_df`.\n",
    "\n",
    "Here's my approach:\n",
    "* Explore the data to understand the different data types present in each column\n",
    "* Determine the appropriate data types for each column\n",
    "* Define the schema using the `StructType` class in Spark and create a `StructField` for each column along with the column name, data type, and other properties\n",
    "* Read the input file using the user-defined schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+------+---+----------+\n",
      "|Emp_No| Emp_Name|Salary|Age|Department|\n",
      "+------+---------+------+---+----------+\n",
      "|   198|   Donald|  2600| 29|        IT|\n",
      "|   199|  Douglas|  2600| 34|     Sales|\n",
      "|   200| Jennifer|  4400| 36| Marketing|\n",
      "|   201|  Michael| 13000| 32|        IT|\n",
      "|   202|      Pat|  6000| 39|        HR|\n",
      "|   203|    Susan|  6500| 36| Marketing|\n",
      "|   204|  Hermann| 10000| 29|   Finance|\n",
      "|   205|  Shelley| 12008| 33|   Finance|\n",
      "|   206|  William|  8300| 37|        IT|\n",
      "|   100|   Steven| 24000| 39|        IT|\n",
      "|   101|    Neena| 17000| 27|     Sales|\n",
      "|   102|      Lex| 17000| 37| Marketing|\n",
      "|   103|Alexander|  9000| 39| Marketing|\n",
      "|   104|    Bruce|  6000| 38|        IT|\n",
      "|   105|    David|  4800| 39|        IT|\n",
      "|   106|    Valli|  4800| 38|     Sales|\n",
      "|   107|    Diana|  4200| 35|     Sales|\n",
      "|   108|    Nancy| 12008| 28|     Sales|\n",
      "|   109|   Daniel|  9000| 35|        HR|\n",
      "|   110|     John|  8200| 31| Marketing|\n",
      "+------+---------+------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Define a Schema for the input data and read the file using the user-defined Schema\n",
    "from pyspark.sql.types import StructType, StructField, IntegerType, StringType\n",
    "\n",
    "schema = StructType([\n",
    "    StructField(\"Emp_No\", IntegerType(), False),\n",
    "    StructField(\"Emp_Name\", StringType(), False),\n",
    "    StructField(\"Salary\", IntegerType(), False),\n",
    "    StructField(\"Age\", IntegerType(), False),\n",
    "    StructField(\"Department\", StringType(), False),\n",
    "])\n",
    "\n",
    "# Create a dataframe on top of a CSV file\n",
    "employees_df = (spark.read\n",
    "    .format(\"csv\")\n",
    "    .schema(schema)\n",
    "    .option(\"header\", \"true\")\n",
    "    .load(\"employees.csv\")\n",
    ")\n",
    "\n",
    "# Display the dataframe content\n",
    "employees_df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 3: Display schema of DataFrame\n",
    "\n",
    "We then display the structure of the `employees_df` DataFrame, including details about all columns and their associated data types.  \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Emp_No: integer (nullable = true)\n",
      " |-- Emp_Name: string (nullable = true)\n",
      " |-- Salary: integer (nullable = true)\n",
      " |-- Age: integer (nullable = true)\n",
      " |-- Department: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Display all columns of the DataFrame, along with their respective data types\n",
    "employees_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 4: Create a temporary view\n",
    "\n",
    "Here, we create a temporary view named `employees` for the `employees_df` DataFrame, enabling Spark SQL queries on the data. \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Create a temporary view named \"employees\" for the DataFrame\n",
    "employees_df.createOrReplaceTempView(\"employees\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 5: Execute an SQL query\n",
    "\n",
    "Next, we compose and execute an SQL query to fetch the records from the `employees` view where the age of employees exceeds 30. Then, display the result of the SQL query, showcasing the filtered records.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----------+------+---+----------+\n",
      "|Emp_No|   Emp_Name|Salary|Age|Department|\n",
      "+------+-----------+------+---+----------+\n",
      "|   199|    Douglas|  2600| 34|     Sales|\n",
      "|   200|   Jennifer|  4400| 36| Marketing|\n",
      "|   201|    Michael| 13000| 32|        IT|\n",
      "|   202|        Pat|  6000| 39|        HR|\n",
      "|   203|      Susan|  6500| 36| Marketing|\n",
      "|   205|    Shelley| 12008| 33|   Finance|\n",
      "|   206|    William|  8300| 37|        IT|\n",
      "|   100|     Steven| 24000| 39|        IT|\n",
      "|   102|        Lex| 17000| 37| Marketing|\n",
      "|   103|  Alexander|  9000| 39| Marketing|\n",
      "|   104|      Bruce|  6000| 38|        IT|\n",
      "|   105|      David|  4800| 39|        IT|\n",
      "|   106|      Valli|  4800| 38|     Sales|\n",
      "|   107|      Diana|  4200| 35|     Sales|\n",
      "|   109|     Daniel|  9000| 35|        HR|\n",
      "|   110|       John|  8200| 31| Marketing|\n",
      "|   111|     Ismael|  7700| 32|        IT|\n",
      "|   112|Jose Manuel|  7800| 34|        HR|\n",
      "|   113|       Luis|  6900| 34|     Sales|\n",
      "|   116|     Shelli|  2900| 37|   Finance|\n",
      "+------+-----------+------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SQL query to fetch solely the records from the View where the age exceeds 30\n",
    "spark.sql(\"SELECT * FROM employees WHERE Age > 30\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 6: Calculate Average Salary by Department\n",
    "\n",
    "Compose an SQL query to retrieve the average salary of employees grouped by department. Then, display the result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 14:======================================================> (73 + 2) / 75]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+\n",
      "|Department|       Avg_Salary|\n",
      "+----------+-----------------+\n",
      "|     Sales|5492.923076923077|\n",
      "|        HR|           5837.5|\n",
      "|   Finance|           5730.8|\n",
      "| Marketing|6633.333333333333|\n",
      "|        IT|           7400.0|\n",
      "+----------+-----------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# SQL query to calculate the average salary of employees grouped by department\n",
    "spark.sql(\n",
    "    \"SELECT Department, AVG(Salary) AS Avg_Salary \"\n",
    "    \"FROM employees \"\n",
    "    \"GROUP BY Department\"\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 7: Filter and Display IT Department Employees\n",
    "\n",
    "We apply a filter to the `employees_df` DataFrame to select records where the department is `'IT'` and display the filtered DataFrame.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+------+---+----------+\n",
      "|Emp_No|Emp_Name|Salary|Age|Department|\n",
      "+------+--------+------+---+----------+\n",
      "|   198|  Donald|  2600| 29|        IT|\n",
      "|   201| Michael| 13000| 32|        IT|\n",
      "|   206| William|  8300| 37|        IT|\n",
      "|   100|  Steven| 24000| 39|        IT|\n",
      "|   104|   Bruce|  6000| 38|        IT|\n",
      "|   105|   David|  4800| 39|        IT|\n",
      "|   111|  Ismael|  7700| 32|        IT|\n",
      "|   129|   Laura|  3300| 38|        IT|\n",
      "|   132|      TJ|  2100| 34|        IT|\n",
      "|   136|   Hazel|  2200| 29|        IT|\n",
      "+------+--------+------+---+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Apply a filter to select records where the department is 'IT'\n",
    "employees_df.filter(employees_df[\"Department\"] == \"IT\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 8: Add 10% Bonus to Salaries\n",
    "\n",
    "Add a new column `\"SalaryAfterBonus\"` to the DataFrame and calculate the new salary by adding a 10% bonus to each employee's salary."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+------+---+----------+------------------+\n",
      "|Emp_No| Emp_Name|Salary|Age|Department|  SalaryAfterBonus|\n",
      "+------+---------+------+---+----------+------------------+\n",
      "|   198|   Donald|  2600| 29|        IT|2860.0000000000005|\n",
      "|   199|  Douglas|  2600| 34|     Sales|2860.0000000000005|\n",
      "|   200| Jennifer|  4400| 36| Marketing|            4840.0|\n",
      "|   201|  Michael| 13000| 32|        IT|14300.000000000002|\n",
      "|   202|      Pat|  6000| 39|        HR| 6600.000000000001|\n",
      "|   203|    Susan|  6500| 36| Marketing| 7150.000000000001|\n",
      "|   204|  Hermann| 10000| 29|   Finance|           11000.0|\n",
      "|   205|  Shelley| 12008| 33|   Finance|13208.800000000001|\n",
      "|   206|  William|  8300| 37|        IT|            9130.0|\n",
      "|   100|   Steven| 24000| 39|        IT|26400.000000000004|\n",
      "|   101|    Neena| 17000| 27|     Sales|           18700.0|\n",
      "|   102|      Lex| 17000| 37| Marketing|           18700.0|\n",
      "|   103|Alexander|  9000| 39| Marketing|            9900.0|\n",
      "|   104|    Bruce|  6000| 38|        IT| 6600.000000000001|\n",
      "|   105|    David|  4800| 39|        IT|            5280.0|\n",
      "|   106|    Valli|  4800| 38|     Sales|            5280.0|\n",
      "|   107|    Diana|  4200| 35|     Sales|            4620.0|\n",
      "|   108|    Nancy| 12008| 28|     Sales|13208.800000000001|\n",
      "|   109|   Daniel|  9000| 35|        HR|            9900.0|\n",
      "|   110|     John|  8200| 31| Marketing|            9020.0|\n",
      "+------+---------+------+---+----------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import col\n",
    "\n",
    "# Add a new column \"SalaryAfterBonus\" with 10% bonus added to the original salary\n",
    "employees_df.withColumn(\"SalaryAfterBonus\", col(\"Salary\") * 1.1).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 9: Find Maximum Salary by Age\n",
    "\n",
    "Group the data by age and calculate the maximum salary for each age group. Then, display the result.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 18:=================================================>   (187 + 12) / 200]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+\n",
      "|Age|Max_Salary|\n",
      "+---+----------+\n",
      "| 26|      3600|\n",
      "| 27|     17000|\n",
      "| 28|     12008|\n",
      "| 29|     10000|\n",
      "| 30|      8000|\n",
      "| 31|      8200|\n",
      "| 32|     13000|\n",
      "| 33|     12008|\n",
      "| 34|      7800|\n",
      "| 35|      9000|\n",
      "| 36|      7900|\n",
      "| 37|     17000|\n",
      "| 38|      6000|\n",
      "| 39|     24000|\n",
      "+---+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import max\n",
    "\n",
    "# Group data by age and calculate the maximum salary for each age group\n",
    "employees_df.groupby([\"Age\"]) \\\n",
    "    .agg(max(\"Salary\").alias(\"Max_Salary\")) \\\n",
    "    .sort(\"Age\") \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 10: Self-Join on Employee Data\n",
    "\n",
    "Next, we join the \"employees_df\" DataFrame with itself based on the \"Emp_No\" column and display the result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+------+---+----------+---------+------+---+----------+\n",
      "|Emp_No| Emp_Name|Salary|Age|Department| Emp_Name|Salary|Age|Department|\n",
      "+------+---------+------+---+----------+---------+------+---+----------+\n",
      "|   198|   Donald|  2600| 29|        IT|   Donald|  2600| 29|        IT|\n",
      "|   199|  Douglas|  2600| 34|     Sales|  Douglas|  2600| 34|     Sales|\n",
      "|   200| Jennifer|  4400| 36| Marketing| Jennifer|  4400| 36| Marketing|\n",
      "|   201|  Michael| 13000| 32|        IT|  Michael| 13000| 32|        IT|\n",
      "|   202|      Pat|  6000| 39|        HR|      Pat|  6000| 39|        HR|\n",
      "|   203|    Susan|  6500| 36| Marketing|    Susan|  6500| 36| Marketing|\n",
      "|   204|  Hermann| 10000| 29|   Finance|  Hermann| 10000| 29|   Finance|\n",
      "|   205|  Shelley| 12008| 33|   Finance|  Shelley| 12008| 33|   Finance|\n",
      "|   206|  William|  8300| 37|        IT|  William|  8300| 37|        IT|\n",
      "|   100|   Steven| 24000| 39|        IT|   Steven| 24000| 39|        IT|\n",
      "|   101|    Neena| 17000| 27|     Sales|    Neena| 17000| 27|     Sales|\n",
      "|   102|      Lex| 17000| 37| Marketing|      Lex| 17000| 37| Marketing|\n",
      "|   103|Alexander|  9000| 39| Marketing|Alexander|  9000| 39| Marketing|\n",
      "|   104|    Bruce|  6000| 38|        IT|    Bruce|  6000| 38|        IT|\n",
      "|   105|    David|  4800| 39|        IT|    David|  4800| 39|        IT|\n",
      "|   106|    Valli|  4800| 38|     Sales|    Valli|  4800| 38|     Sales|\n",
      "|   107|    Diana|  4200| 35|     Sales|    Diana|  4200| 35|     Sales|\n",
      "|   108|    Nancy| 12008| 28|     Sales|    Nancy| 12008| 28|     Sales|\n",
      "|   109|   Daniel|  9000| 35|        HR|   Daniel|  9000| 35|        HR|\n",
      "|   110|     John|  8200| 31| Marketing|     John|  8200| 31| Marketing|\n",
      "+------+---------+------+---+----------+---------+------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Join the DataFrame with itself based on the \"Emp_No\" column\n",
    "employees_df.join(employees_df, \"Emp_No\", \"inner\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 11: Calculate Average Employee Age\n",
    "\n",
    "Calculate the average age of employees using the built-in aggregation function and display the result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+\n",
      "|Avg_Age|\n",
      "+-------+\n",
      "|  33.56|\n",
      "+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Calculate the average age of employees\n",
    "from pyspark.sql.functions import avg\n",
    "\n",
    "employees_df.agg(avg(\"Age\").alias(\"Avg_Age\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 12: Calculate Total Salary by Department\n",
    "\n",
    "Calculate the total salary for each department using the built-in aggregation function and display the result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+\n",
      "|Department|Total_Salary|\n",
      "+----------+------------+\n",
      "|     Sales|       71408|\n",
      "|        HR|       46700|\n",
      "|   Finance|       57308|\n",
      "| Marketing|       59700|\n",
      "|        IT|       74000|\n",
      "+----------+------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Calculate the total salary for each department.\n",
    "# Hint - Use GroupBy and Aggregate functions\n",
    "from pyspark.sql.functions import sum \n",
    "\n",
    "employees_df.groupBy(\"Department\") \\\n",
    "    .agg(sum(\"Salary\") \\\n",
    "    .alias(\"Total_Salary\")) \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 13: Sort Data by Age and Salary\n",
    "\n",
    "Sort the DataFrame by age in ascending order and then by salary in descending order. Then, display the sorted DataFrame.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+------+---+----------+\n",
      "|Emp_No| Emp_Name|Salary|Age|Department|\n",
      "+------+---------+------+---+----------+\n",
      "|   137|   Renske|  3600| 26| Marketing|\n",
      "|   101|    Neena| 17000| 27|     Sales|\n",
      "|   114|      Den| 11000| 27|   Finance|\n",
      "|   108|    Nancy| 12008| 28|     Sales|\n",
      "|   130|    Mozhe|  2800| 28| Marketing|\n",
      "|   126|    Irene|  2700| 28|        HR|\n",
      "|   204|  Hermann| 10000| 29|   Finance|\n",
      "|   115|Alexander|  3100| 29|   Finance|\n",
      "|   134|  Michael|  2900| 29|     Sales|\n",
      "|   198|   Donald|  2600| 29|        IT|\n",
      "|   140|   Joshua|  2500| 29|   Finance|\n",
      "|   136|    Hazel|  2200| 29|        IT|\n",
      "|   120|  Matthew|  8000| 30|        HR|\n",
      "|   110|     John|  8200| 31| Marketing|\n",
      "|   127|    James|  2400| 31|        HR|\n",
      "|   201|  Michael| 13000| 32|        IT|\n",
      "|   111|   Ismael|  7700| 32|        IT|\n",
      "|   119|    Karen|  2500| 32|   Finance|\n",
      "|   205|  Shelley| 12008| 33|   Finance|\n",
      "|   124|    Kevin|  5800| 33| Marketing|\n",
      "+------+---------+------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Sort the DataFrame by age in ascending order and then by salary\n",
    "# in descending order\n",
    "employees_df.sort([\"Age\", \"Salary\"], ascending=[True, False]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 14: Count Employees in Each Department\n",
    "\n",
    "Calculate the number of employees in each department and display the result.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+\n",
      "|Department|Emp_Count|\n",
      "+----------+---------+\n",
      "|     Sales|       13|\n",
      "|        HR|        8|\n",
      "|   Finance|       10|\n",
      "| Marketing|        9|\n",
      "|        IT|       10|\n",
      "+----------+---------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import count\n",
    "\n",
    "# Calculate the number of employees in each department\n",
    "employees_df.groupBy(\"Department\") \\\n",
    "    .agg(count(\"*\").alias(\"Emp_Count\")) \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 15: Filter Employees with the letter 'o' in the Name\n",
    "\n",
    "Apply a filter to select records where the employee’s name contains the letter `‘o’` and display the filtered DataFrame.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----------+------+---+----------+\n",
      "|Emp_No|   Emp_Name|Salary|Age|Department|\n",
      "+------+-----------+------+---+----------+\n",
      "|   198|     Donald|  2600| 29|        IT|\n",
      "|   199|    Douglas|  2600| 34|     Sales|\n",
      "|   110|       John|  8200| 31| Marketing|\n",
      "|   112|Jose Manuel|  7800| 34|        HR|\n",
      "|   130|      Mozhe|  2800| 28| Marketing|\n",
      "|   133|      Jason|  3300| 38|     Sales|\n",
      "|   139|       John|  2700| 36|     Sales|\n",
      "|   140|     Joshua|  2500| 29|   Finance|\n",
      "+------+-----------+------+---+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Apply a filter to select records where the employee's name\n",
    "# contains the letter 'o'\n",
    "employees_df.filter(col(\"Emp_Name\").contains(\"o\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Stop Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Change Log\n",
    "\n",
    "|Date (YYYY-MM-DD)|Version|Changed By|Change Description|\n",
    "|-|-|-|-|\n",
    "|2023-09-01|0.1|Lavanya T S|Initial version|\n",
    "|2023-09-11|0.2|Pornima More|QA pass with edits|\n",
    "|2024-04-29|0.3|Pravin Regismond|Modified to fulfill the project requirements|"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copyright © 2023 IBM Corporation. All rights reserved."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python",
   "language": "python",
   "name": "conda-env-python-py"
  },
  "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}