PracticeProject.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<p style=\"text-align:center\">\n",
    "    <a href=\"https://skills.network\" target=\"_blank\">\n",
    "    <img src=\"https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/assets/logos/SN_web_lightmode.png\" width=\"200\" alt=\"Skills Network Logo\">\n",
    "    </a>\n",
    "</p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Practice Project\n",
    "\n",
    "Estimated time needed: **60** minutes\n",
    "\n",
    "This practice project focuses on data transformation and integration using PySpark. You will work with two datasets, perform various transformations such as adding columns, renaming columns, dropping unnecessary columns, joining dataframes, and finally, writing the results into both a Hive warehouse and an HDFS file system.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Prerequisites \n",
    "\n",
    "For this lab assignment, you will use wget, Python and Spark (PySpark). Therefore, it's essential to make sure that the below-specified libraries are installed in your lab environment or within Skills Network (SN) Labs.  \n",
    "\n",
    " \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: wget in /home/jupyterlab/conda/envs/python/lib/python3.7/site-packages (3.2)\n",
      "Requirement already satisfied: pyspark in /home/jupyterlab/conda/envs/python/lib/python3.7/site-packages (3.4.3)\n",
      "Requirement already satisfied: findspark in /home/jupyterlab/conda/envs/python/lib/python3.7/site-packages (2.0.1)\n",
      "Requirement already satisfied: py4j==0.10.9.7 in /home/jupyterlab/conda/envs/python/lib/python3.7/site-packages (from pyspark) (0.10.9.7)\n"
     ]
    }
   ],
   "source": [
    "# Installing required packages\n",
    "\n",
    "!pip install wget pyspark  findspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Prework - Initiate the Spark Session\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import findspark\n",
    "\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# PySpark is the Spark API for Python. In this lab, we use PySpark to initialize the SparkContext.   \n",
    "\n",
    "from pyspark import SparkContext, SparkConf\n",
    "\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Creating a SparkContext object\n",
    "\n",
    "sc = SparkContext.getOrCreate()\n",
    "\n",
    "# Creating a Spark Session\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark DataFrames basic example\") \\\n",
    "    .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 1: Load datasets into PySpark DataFrames\n",
    "\n",
    "Download the datasets from the folloing links using `wget` and load it in a Spark Dataframe.\n",
    "\n",
    "1. https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/IBM-BD0225EN-SkillsNetwork/labs/data/dataset1.csv  \n",
    "2. https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/IBM-BD0225EN-SkillsNetwork/labs/data/dataset2.csv  \n",
    "\n",
    "*Hint: Import wget*\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'dataset2 (1).csv'"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# download the dataset using wget\n",
    "import wget\n",
    "\n",
    "link_to_data1 = 'https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/IBM-BD0225EN-SkillsNetwork/labs/data/dataset1.csv'\n",
    "wget.download(link_to_data1)\n",
    "\n",
    "link_to_data2 = 'https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/IBM-BD0225EN-SkillsNetwork/labs/data/dataset2.csv'\n",
    "wget.download(link_to_data2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "# download the dataset using wget\n",
    "import wget\n",
    "\n",
    "link_to_data1 = 'https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/IBM-BD0225EN-SkillsNetwork/labs/data/dataset1.csv'\n",
    "wget.download(link_to_data1)\n",
    "\n",
    "link_to_data2 = 'https://cf-courses-data.s3.us.cloud-object-storage.appdomain.cloud/IBM-BD0225EN-SkillsNetwork/labs/data/dataset2.csv'\n",
    "wget.download(link_to_data2)\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "#load the data into a pyspark dataframe\n",
    "\n",
    "df1 = spark.read.csv(\"dataset1.csv\", header=True, inferSchema=True)\n",
    "df2 = spark.read.csv(\"dataset2.csv\", header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#load the data into a pyspark dataframe\n",
    "    \n",
    "df1 = spark.read.csv(\"dataset1.csv\", header=True, inferSchema=True)\n",
    "df2 = spark.read.csv(\"dataset2.csv\", header=True, inferSchema=True)\n",
    "```\n",
    "\n",
    "</details>\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 2: Display the schema of both dataframes\n",
    "\n",
    "Display the schema of `df1` and `df2` to understand the structure of the datasets.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customer_id: integer (nullable = true)\n",
      " |-- date_column: string (nullable = true)\n",
      " |-- amount: integer (nullable = true)\n",
      " |-- description: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      "\n",
      "root\n",
      " |-- customer_id: integer (nullable = true)\n",
      " |-- transaction_date: string (nullable = true)\n",
      " |-- value: integer (nullable = true)\n",
      " |-- notes: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#print the schema of df1 and df2\n",
    "\n",
    "df1.printSchema()\n",
    "df2.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#print the schema of df1 and df2\n",
    "    \n",
    "df1.printSchema()\n",
    "df2.printSchema()\n",
    "```\n",
    "\n",
    "</details>\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 3: Add a new column to each dataframe\n",
    "\n",
    "Add a new column named **year** to `df1` and **quarter** to `df2` representing the year and quarter of the data.\n",
    "\n",
    "*Hint: use withColumn. Convert the date columns which are present as string to date before extracting the year and quarter information*\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import year, quarter, to_date\n",
    "\n",
    "#Add new column year to df1\n",
    "df1 = df1.withColumn('year', year(to_date('date_column','dd/MM/yyyy')))\n",
    "    \n",
    "#Add new column quarter to df2    \n",
    "df2 = df2.withColumn('quarter', quarter(to_date('transaction_date','dd/MM/yyyy')))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "from pyspark.sql.functions import year, quarter\n",
    "\n",
    "#Add new column year to df1\n",
    "df1 = df1.withColumn('year', year(to_date('date_column','dd/MM/yyyy')))\n",
    "    \n",
    "#Add new column quarter to df2    \n",
    "df2 = df2.withColumn('quarter', quarter(to_date('transaction_date','dd/MM/yyyy')))```\n",
    "\n",
    "</details>\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 4: Rename columns in both dataframes\n",
    "\n",
    "Rename the column **amount** to **transaction_amount** in `df1` and **value** to **transaction_value** in `df2`.\n",
    "\n",
    "*Hint: Use withColumnRenamed*\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Rename df1 column amount to transaction_amount\n",
    "df1 = df1.withColumnRenamed('amount', 'transaction_amount')\n",
    "    \n",
    "#Rename df2 column value to transaction_value\n",
    "df2 = df2.withColumnRenamed('value', 'transaction_value')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#Rename df1 column amount to transaction_amount\n",
    "df1 = df1.withColumnRenamed('amount', 'transaction_amount')\n",
    "    \n",
    "#Rename df2 column value to transaction_value\n",
    "df2 = df2.withColumnRenamed('value', 'transaction_value')\n",
    "```\n",
    "\n",
    "</details>\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 5: Drop unnecessary columns\n",
    "\n",
    "Drop the columns **description** and **location** from `df1` and **notes** from `df2`.\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Drop columns description and location from df1\n",
    "df1 = df1.drop('description', 'location')\n",
    "    \n",
    "#Drop column notes from df2\n",
    "df2 = df2.drop('notes')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#Drop columns description and location from df1\n",
    "df1 = df1.drop('description', 'location')\n",
    "    \n",
    "#Drop column notes from df2\n",
    "df2 = df2.drop('notes')\n",
    "```\n",
    "\n",
    "</details>\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 6: Join dataframes based on a common column\n",
    "\n",
    "Join `df1` and `df2` based on the common column **customer_id** and create a new dataframe named `joined_df`.\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "#join df1 and df2 based on common column customer_id\n",
    "joined_df = df1.join(df2, 'customer_id', 'inner')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#join df1 and df2 based on common column customer_id\n",
    "joined_df = df1.join(df2, 'customer_id', 'inner')\n",
    "    \n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 7: Filter data based on a condition\n",
    "\n",
    "Filter `joined_df` to include only transactions where \"transaction_amount\" is greater than 1000 and create a new dataframe named `filtered_df`.\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "# filter the dataframe for transaction amount > 1000\n",
    "filtered_df = joined_df.filter(\"transaction_amount > 1000\") "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "# filter the dataframe for transaction amount > 1000\n",
    "filtered_df = joined_df.filter(\"transaction_amount > 1000\")    \n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 8: Aggregate data by customer\n",
    "\n",
    "Calculate the total transaction amount for each customer in `filtered_df` and display the result.\n",
    "\n",
    "*Hint: Use sum from pyspark.sql.functions*\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 25:================================>                       (35 + 9) / 61]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------------+\n",
      "|customer_id|total_amount|\n",
      "+-----------+------------+\n",
      "|         31|        3200|\n",
      "|         85|        1800|\n",
      "|         78|        1500|\n",
      "|         34|        1200|\n",
      "|         81|        5500|\n",
      "|         28|        2600|\n",
      "|         76|        2600|\n",
      "|         27|        4200|\n",
      "|         91|        3200|\n",
      "|         22|        1200|\n",
      "|         93|        5500|\n",
      "|          1|        5000|\n",
      "|         52|        2600|\n",
      "|         13|        4800|\n",
      "|          6|        4500|\n",
      "|         16|        2600|\n",
      "|         40|        2600|\n",
      "|         94|        1200|\n",
      "|         57|        5500|\n",
      "|         54|        1500|\n",
      "+-----------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import sum\n",
    "   \n",
    "# group by customer_id and aggregate the sum of transaction amount\n",
    "\n",
    "total_amount_per_customer = filtered_df.groupBy('customer_id').agg(sum('transaction_amount').alias('total_amount'))\n",
    "\n",
    "#display the result\n",
    "total_amount_per_customer.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "from pyspark.sql.functions import sum\n",
    "   \n",
    "# group by customer_id and aggregate the sum of transaction amount\n",
    "\n",
    "total_amount_per_customer = filtered_df.groupBy('customer_id').agg(sum('transaction_amount').alias('total_amount'))\n",
    "\n",
    "#display the result\n",
    "total_amount_per_customer.show()\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 9: Write the result to a Hive table\n",
    "\n",
    "Write `total_amount_per_customer` to a Hive table named **customer_totals**.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Write total_amount_per_customer to a Hive table named customer_totals\n",
    "total_amount_per_customer.write.mode(\"overwrite\").saveAsTable(\"customer_totals\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "# Write total_amount_per_customer to a Hive table named customer_totals\n",
    "total_amount_per_customer.write.mode(\"overwrite\").saveAsTable(\"customer_totals\")\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 10: Write the filtered data to HDFS\n",
    "\n",
    "Write `filtered_df` to HDFS in parquet format to a file named **filtered_data**.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Write filtered_df to HDFS in parquet format file filtered_data.parquet\n",
    "\n",
    "filtered_df.write.mode(\"overwrite\").parquet(\"filtered_data.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#Write filtered_df to HDFS in parquet format file filtered_data\n",
    "\n",
    "filtered_df.write.mode(\"overwrite\").parquet(\"filtered_data.parquet\")\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 11: Add a new column based on a condition\n",
    "\n",
    "Add a new column named **high_value** to `df1` indicating whether the transaction_amount is greater than 5000. When the value is greater than 5000, the value of the column should be **Yes**. When the value is less than or equal to 5000, the value of the column should be **No**. \n",
    "\n",
    "*Hint: Use when and lit from pyspark.sql.functions\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add new column with value indicating whether transaction amount is > 5000 or not\n",
    "from pyspark.sql.functions import when, lit\n",
    "\n",
    "# Add new column with value indicating whether transaction amount is > 5000 or not\n",
    "df1 = df1.withColumn(\"high_value\", when(df1.transaction_amount > 5000, lit(\"Yes\")).otherwise(lit(\"No\")))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "from pyspark.sql.functions import when, lit\n",
    "\n",
    "# Add new column with value indicating whether transaction amount is > 5000 or not\n",
    "df1 = df1.withColumn(\"high_value\", when(df1.transaction_amount > 5000, lit(\"Yes\")).otherwise(lit(\"No\")))\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 12: Calculate the average transaction value per quarter\n",
    "\n",
    "Calculate and display the average transaction value for each quarter in `df2` and create a new dataframe named `average_value_per_quarter` with column `avg_trans_val`.\n",
    "\n",
    "*Hint: Use avg from pyspark.sql.functions*\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 40:=======================================>                (53 + 8) / 75]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|quarter|     avg_trans_val|\n",
      "+-------+------------------+\n",
      "|      1| 1111.111111111111|\n",
      "|      3|1958.3333333333333|\n",
      "|      4| 816.6666666666666|\n",
      "|      2|            1072.0|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import avg\n",
    "\n",
    "#calculate the average transaction value for each quarter in df2\n",
    "average_value_per_quarter = df2.groupBy('quarter').agg(avg(\"transaction_value\").alias(\"avg_trans_val\"))\n",
    "\n",
    "    \n",
    "#show the average transaction value for each quarter in df2    \n",
    "average_value_per_quarter.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "from pyspark.sql.functions import avg\n",
    "\n",
    "#calculate the average transaction value for each quarter in df2\n",
    "average_value_per_quarter = df2.groupBy('quarter').agg(avg(\"transaction_value\").alias(\"avg_trans_val\"))\n",
    "\n",
    "    \n",
    "#show the average transaction value for each quarter in df2    \n",
    "average_value_per_quarter.show()\n",
    "\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 13: Write the result to a Hive table\n",
    "\n",
    "Write `average_value_per_quarter` to a Hive table named **quarterly_averages**.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Write average_value_per_quarter to a Hive table named quarterly_averages\n",
    "\n",
    "average_value_per_quarter.write.mode(\"overwrite\").saveAsTable(\"quarterly_averages\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#Write average_value_per_quarter to a Hive table named quarterly_averages\n",
    "\n",
    "average_value_per_quarter.write.mode(\"overwrite\").saveAsTable(\"quarterly_averages\")\n",
    "\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 14: Calculate the total transaction value per year\n",
    "\n",
    "Calculate and display the total transaction value for each year in `df1` and create a new dataframe named `total_value_per_year` with column `total_transaction_val`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---------------------+\n",
      "|year|total_transaction_val|\n",
      "+----+---------------------+\n",
      "|2025|                25700|\n",
      "|2027|                25700|\n",
      "|2023|                28100|\n",
      "|2022|                29800|\n",
      "|2026|                25700|\n",
      "|2029|                25700|\n",
      "|2030|                 9500|\n",
      "|2028|                25700|\n",
      "|2024|                25700|\n",
      "+----+---------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# calculate the total transaction value for each year in df1.\n",
    "total_value_per_year = df1.groupBy('year').agg(sum(\"transaction_amount\").alias(\"total_transaction_val\"))\n",
    "\n",
    "# show the total transaction value for each year in df1.\n",
    "total_value_per_year.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "# calculate the total transaction value for each year in df1.\n",
    "total_value_per_year = df1.groupBy('year').agg(sum(\"transaction_amount\").alias(\"total_transaction_val\"))\n",
    "\n",
    "# show the total transaction value for each year in df1.\n",
    "total_value_per_year.show()\n",
    "\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Task 15: Write the result to HDFS\n",
    "\n",
    "Write `total_value_per_year` to HDFS in the CSV format to file named **total_value_per_year**.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Write total_value_per_year to HDFS in the CSV format\n",
    "\n",
    "total_value_per_year.write.mode(\"overwrite\").csv(\"total_value_per_year.csv\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<details>\n",
    "    <summary>Click here for Solution</summary>\n",
    "\n",
    "```python\n",
    "\n",
    "#Write total_value_per_year to HDFS in the CSV format\n",
    "\n",
    "total_value_per_year.write.mode(\"overwrite\").csv(\"total_value_per_year.csv\")\n",
    "\n",
    "```\n",
    "\n",
    "</details>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Show the number of files in the notebook:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['.ipynb_checkpoints',\n",
       " 'DataFrames.ipynb',\n",
       " 'PracticeProject.ipynb',\n",
       " 'SparkIntro.ipynb',\n",
       " 'SparkSQL.ipynb',\n",
       " 'dataset1 (1).csv',\n",
       " 'dataset1.csv',\n",
       " 'dataset2 (1).csv',\n",
       " 'dataset2.csv',\n",
       " 'filtered_data.parquet',\n",
       " 'people.json',\n",
       " 'people2.json',\n",
       " 'spark-warehouse',\n",
       " 'total_value_per_year.csv']"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import os\n",
    "os.listdir()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read Hive table into pandas dataframe:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   quarter  avg_trans_val\n",
      "0        1    1111.111111\n",
      "1        3    1958.333333\n",
      "2        4     816.666667\n",
      "3        2    1072.000000\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "from pyspark import SparkConf, SparkContext\n",
    "\n",
    "# Read Hive table into a Spark DataFrame\n",
    "df = spark.read.table('quarterly_averages')\n",
    "\n",
    "# Convert Spark DataFrame to Pandas DataFrame\n",
    "pandas_df = df.toPandas()\n",
    "\n",
    "# Process the rows using Pandas\n",
    "# ...\n",
    "\n",
    "# Example: Display the first 5 rows\n",
    "print(pandas_df.head(5))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Congratulations! You have completed the lab.\n",
    "This practice project provides hands-on experience with data transformation and integration using PySpark. You've performed various tasks, including adding columns, renaming columns, dropping unnecessary columns, joining dataframes, and writing the results into both a Hive warehouse and an HDFS file system.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Authors\n",
    "\n",
    "Raghul Ramesh\n",
    "\n",
    "Lavanya T S\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<!--## Change Log -->\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<!--|Date (YYYY-MM-DD)|Version|Changed By|Change Description|\n",
    "|-|-|-|-|\n",
    "|2023-09-01|0.1|Lavanya T S|Initial version|\n",
    "|2023-09-08|0.2|Pornima More|QA pass with edits|-->\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3 align=\"center\"> &#169; IBM Corporation. All rights reserved. <h3/>\n"
   ]
  }
 ],
 "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"
  },
  "prev_pub_hash": "11f72aae77d8016a27f044c7c8c6c15789e27d81d5809ee65ea885d8da7126a7"
 },
 "nbformat": 4,
 "nbformat_minor": 4
}