Task1_data_ingestion/Data_ingestion_student_version.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Processing Big Data - Data Ingestion\n",
    "© Explore Data Science Academy\n",
    "\n",
    "## Honour Code\n",
    "I {**YOUR NAME**, **YOUR SURNAME**}, confirm - by submitting this document - that the solutions in this notebook are a result of my own work and that I abide by the [EDSA honour code](https://drive.google.com/file/d/1QDCjGZJ8-FmJE3bZdIQNwnJyQKPhHZBn/view?usp=sharing).\n",
    "    Non-compliance with the honour code constitutes a material breach of contract.\n",
    "\n",
    "\n",
    "\n",
    "## Context \n",
    "\n",
    "To work constructively with any dataset, one needs to create an ingestion profile to make sure that the data at the source can be readily consumed. For this section of the predict, as the Data Engineer in the team, you will be required to design and implement the ingestion process. For the purposes of the project the AWS cloud storage service, namely, the S3 bucket service will act as your data source. All the data required can be found [here](https://processing-big-data-predict-stocks-data.s3.eu-west-1.amazonaws.com/stocks.zip).\n",
    "\n",
    "<div align=\"center\" style=\"width: 600px; font-size: 80%; text-align: center; margin: 0 auto\">\n",
    "<img src=\"https://raw.githubusercontent.com/Explore-AI/Pictures/master/data_engineering/transform/predict/DataIngestion.jpg\"\n",
    "     alt=\"Data Ingestion\"\n",
    "     style=\"float: center; padding-bottom=0.5em\"\n",
    "     width=40%/>\n",
    "     <p><em>Figure 1. Data Ingestion</em></p>\n",
    "</div>\n",
    "\n",
    "Your manager, Gnissecorp Atadgib, knowing very well that you've recently completed your Data Engineering qualification, asks you to make use of Apache Spark for the ingestion as well as the rest of the project. His rationale being, that stock market data is generated every day and is quite time-sensitive and would require scalability when deploying to a production environment. \n",
    "\n",
    "## Dataset - US Nasdaq\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "<div align=\"center\" style=\"width: 600px; font-size: 80%; text-align: center; margin: 0 auto\">\n",
    "<img src=\"https://raw.githubusercontent.com/Explore-AI/Pictures/master/data_engineering/transform/predict/Nasdaq.png\"\n",
    "     alt=\"Nasdaq\"\n",
    "     style=\"float: center; padding-bottom=0.5em\"\n",
    "     width=50%/>\n",
    "     <p><em>Figure 2. Nasdaq</em></p>\n",
    "</div>\n",
    "\n",
    "The data that you will be working with is a historical snapshot of market data taken from the Nasdaq electronic market. This dataset contains historical daily prices for all tickers currently trading on Nasdaq. The up-to-date list can be found on their [website](https://www.nasdaq.com/)\n",
    "\n",
    "\n",
    "The provided data contains price data dating back from 02 January 1962 up until 01 April 2020. The data found in the S3 bucket has been stored in the following structure:\n",
    "\n",
    "```\n",
    "     stocks/<Year>/<Month>/<Day>/stocks.csv\n",
    "```\n",
    "Each CSV file for every trading day contains the following details:\n",
    "- **Date** - specifies trading date\n",
    "- **Open** - opening price\n",
    "- **High** - maximum price during the day\n",
    "- **Low** - minimum price during the day\n",
    "- **Close** - close price adjusted for splits\n",
    "- **Adj Close** - close price adjusted for both dividends and splits\n",
    "- **Volume** - the number of shares that changed hands during a given day"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Basic initialisation\n",
    "To get you started, let's import some basic Python libraries as well as Spark modules and functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Remember that we need a `SparkContext` and `SparkSession` to interface with Spark.\n",
    "We will mostly be using the `SparkContext` to interact with RDDs and the `SparkSession` to interface with Python objects.\n",
    "\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    ">Initialise a new **Spark Context** and **Session** that you will use to interface with Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Investigate dataset schema\n",
    "At this point, it is enough to read in a single file to ascertain the data structure. You will be required to use the information obtained from the small subset to create a data schema. This data schema will be used when reading the entire dataset using Spark.\n",
    "\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    ">Make use of Pandas to read in a single file and investigate the plausible data types to be used when creating a Spark data schema. \n",
    ">\n",
    ">*You may use as many coding cells as necessary.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read CSV files\n",
    "\n",
    "When working with big data, it is often not tenable to keep processing an entire data batch when you are in the process of development - this can be quite time-consuming. If the data is uniform, it is sufficient to work with a smaller subset to create basic functionality. Your manager has identified the year **1962** to perform the initial testing for data ingestion. \n",
    "\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    ">Read in the data for **1962** using a data schema that purely uses string data types. You will be required to convert to the appropriate data types at a later stage.\n",
    ">\n",
    ">*You may use as many coding cells as necessary.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Update column names\n",
    "To make the data easier to work with, you will need to make a few changes:\n",
    "1. Column headers should all be in lowercase; and\n",
    "2. Whitespaces should be replaced with underscores.\n",
    "\n",
    "\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    ">Make sure that the column headers are all in lowercase and that any whitespaces are replaced with underscores.\n",
    ">\n",
    ">*You may use as many coding cells as necessary.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Null Values\n",
    "Null values often represent missing pieces of data. It is always good to know where your null values lie - so you can quickly identify and remedy any issues stemming from these.\n",
    "\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    ">Write code to count the number of null values found in each column.\n",
    ">\n",
    ">*You may use as many coding cells as necessary.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data type conversion - The final data schema\n",
    "\n",
    "Now that we have identified the number of missing values in the data set, we'll move on to convert our data schema to the required data types. \n",
    "\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    ">Use typecasting to convert the string data types in your current data schema to more appropriate data types.\n",
    ">\n",
    ">*You may use as many coding cells as necessary.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Consolidate missing values\n",
    "We have to check if the data type conversion above was done correctly.\n",
    "If the casting was not successful, a null value gets inserted into the dataframe. You can thus check for successful conversion by determining if any null values are included in the resulting dataframe.\n",
    "\n",
    "\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    ">Write code to compare the number of invalid entries (nulls) pre-conversion and post-conversion.\n",
    ">\n",
    ">*You may use as many coding cells as necessary.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here you should be able to see if any of your casts went wrong. \n",
    "Do not attempt to correct any missing values at this point. This will be dealt with in later sections of the predict."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Generate parquet files\n",
    "When writing in Spark, we typically use parquet format. This format allows parallel writing using Spark's optimisation while maintaining other useful things like metadata.\n",
    "\n",
    "When writing, it is good to make sure that the data is sufficiently partitioned. \n",
    "\n",
    "Generally, data should be partitioned with one partition for every 200MB of data, but this also depends on the size of your cluster and executors. \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check the size of the dataframe before partitioning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.serializers import PickleSerializer, AutoBatchedSerializer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd = df.rdd._reserialize(AutoBatchedSerializer(PickleSerializer()))\n",
    "obj = rdd.ctx._jvm.org.apache.spark.mllib.api.python.SerDe.pythonToJava(rdd._jrdd, True)\n",
    "size = sc._jvm.org.apache.spark.util.SizeEstimator.estimate(obj)\n",
    "size_MB = size/1000000\n",
    "partitions = max(int(size_MB/200), 2)\n",
    "print(f'The dataframe is {size_MB} MB')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Write parquet files to the local directory\n",
    "> ℹ️ **Instructions** ℹ️\n",
    ">\n",
    "> Use the **coalesce** function and the number of **partitions** derived above to write parquet files to your local directory \n",
    ">\n",
    ">*You may use as many coding cells as necessary.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TODO: Write your code here"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Raw Cell Format",
  "interpreter": {
   "hash": "24a0a2ddc4dffcb168e507551dd24967ddc40ea2df7a72a200a74e0ae6d88beb"
  },
  "kernelspec": {
   "display_name": "Python 3.8.11 64-bit ('spark': conda)",
   "name": "python3"
  },
  "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.8.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}