SparkETLCode.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ETL Project (Spar Nord Bank ATM Data Mart) by Priya Mishra"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Importing necessary libraries and setting up SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "os.environ[\"PYSPARK_PYTHON\"] = \"/opt/cloudera/parcels/Anaconda/bin/python\"\n",
    "os.environ[\"JAVA_HOME\"] = \"/usr/java/jdk1.8.0_232-cloudera/jre\"\n",
    "os.environ[\"SPARK_HOME\"]=\"/opt/cloudera/parcels/SPARK2-2.3.0.cloudera2-1.cdh5.13.3.p0.316101/lib/spark2/\"\n",
    "os.environ[\"PYLIB\"] = os.environ[\"SPARK_HOME\"] + \"/python/lib\"\n",
    "sys.path.insert(0, os.environ[\"PYLIB\"] +\"/py4j-0.10.6-src.zip\")\n",
    "sys.path.insert(0, os.environ[\"PYLIB\"] +\"/pyspark.zip\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://ip-10-0-0-166.ec2.internal:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.0.cloudera2</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>jupyter_Spark</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f47f774f190>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('jupyter_Spark').master(\"local\").getOrCreate()\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Reading the data to Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Reading the data from HDFS, using default schema at first, just to make sure data is getting loaded"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(\"/user/root/SRC_ATM_TRANS/part-m-00000\", header = False, inferSchema = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+---+------+---+------+---+---+----------+-----------+----+----+------+------+----+----------+----+----------+----+----+-----+------+-------+--------+------+----+----+----+----+-----+----+----+----+----------+\n",
      "| _c0|    _c1|_c2|   _c3|_c4|   _c5|_c6|_c7|       _c8|        _c9|_c10|_c11|  _c12|  _c13|_c14|      _c15|_c16|      _c17|_c18|_c19| _c20|  _c21|   _c22|    _c23|  _c24|_c25|_c26|_c27|_c28| _c29|_c30|_c31|_c32|      _c33|\n",
      "+----+-------+---+------+---+------+---+---+----------+-----------+----+----+------+------+----+----------+----+----------+----+----+-----+------+-------+--------+------+----+----+----+----+-----+----+----+----+----------+\n",
      "|2017|January|  1|Sunday|  0|Active|  1|NCR|NÃƒÂ¦stved|Farimagsvej|   8|4700|55.233|11.763| DKK|MasterCard|5643|Withdrawal|null|null|55.23|11.761|2616038|Naestved|281.15|1014|  87|   7| 260|0.215|  92| 500|Rain|light rain|\n",
      "+----+-------+---+------+---+------+---+---+----------+-----------+----+----+------+------+----+----------+----+----------+----+----+-----+------+-------+--------+------+----+----+----+----+-----+----+----+----+----------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "####  Creating custom input schema using StrucType and reading the data "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType, DoubleType, LongType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "Schema = StructType([StructField('year', IntegerType(), nullable = True),\n",
    "                        StructField('month', StringType(), True),\n",
    "                        StructField('day', IntegerType(), True),\n",
    "                        StructField('weekday', StringType(), True),\n",
    "                        StructField('hour', IntegerType(), True),\n",
    "                        StructField('atm_status', StringType(), True),\n",
    "                        StructField('atm_id', StringType(), True),\n",
    "                        StructField('atm_manufacturer', StringType(), True),\n",
    "                        StructField('atm_location', StringType(), True),\n",
    "                        StructField('atm_streetname', StringType(), True),\n",
    "                        StructField('atm_street_number', IntegerType(), True),\n",
    "                        StructField('atm_zipcode', IntegerType(), True),\n",
    "                        StructField('atm_lat', DoubleType(), True),\n",
    "                        StructField('atm_lon', DoubleType(), True),\n",
    "                        StructField('currency', StringType(), True),\n",
    "                        StructField('card_type', StringType(), True),\n",
    "                        StructField('transaction_amount', IntegerType(), True),\n",
    "                        StructField('service', StringType(), True),\n",
    "                        StructField('message_code', StringType(), True),\n",
    "                        StructField('message_text', StringType(), True),\n",
    "                        StructField('weather_lat', DoubleType(), True),\n",
    "                        StructField('weather_lon', DoubleType(), True),\n",
    "                        StructField('weather_city_id', IntegerType(), True),\n",
    "                        StructField('weather_city_name', StringType(), True),\n",
    "                        StructField('temp', DoubleType(), True),\n",
    "                        StructField('pressure', IntegerType(), True),\n",
    "                        StructField('humidity', IntegerType(), True),\n",
    "                        StructField('wind_speed', IntegerType(), True),\n",
    "                        StructField('wind_deg', IntegerType(), True),\n",
    "                        StructField('rain_3h', DoubleType(), True),\n",
    "                        StructField('clouds_all', IntegerType(), True),\n",
    "                        StructField('weather_id', IntegerType(), True),\n",
    "                        StructField('weather_main', StringType(), True),\n",
    "                        StructField('weather_description', StringType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(\"/user/root/SRC_ATM_TRANS/part-m-00000\", header = False, schema = Schema)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Verifying the count of the records loaded into the Dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2468572"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('*').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Taking a glance at the Schema, Dataframe and the Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- year: integer (nullable = true)\n",
      " |-- month: string (nullable = true)\n",
      " |-- day: integer (nullable = true)\n",
      " |-- weekday: string (nullable = true)\n",
      " |-- hour: integer (nullable = true)\n",
      " |-- atm_status: string (nullable = true)\n",
      " |-- atm_id: string (nullable = true)\n",
      " |-- atm_manufacturer: string (nullable = true)\n",
      " |-- atm_location: string (nullable = true)\n",
      " |-- atm_streetname: string (nullable = true)\n",
      " |-- atm_street_number: integer (nullable = true)\n",
      " |-- atm_zipcode: integer (nullable = true)\n",
      " |-- atm_lat: double (nullable = true)\n",
      " |-- atm_lon: double (nullable = true)\n",
      " |-- currency: string (nullable = true)\n",
      " |-- card_type: string (nullable = true)\n",
      " |-- transaction_amount: integer (nullable = true)\n",
      " |-- service: string (nullable = true)\n",
      " |-- message_code: string (nullable = true)\n",
      " |-- message_text: string (nullable = true)\n",
      " |-- weather_lat: double (nullable = true)\n",
      " |-- weather_lon: double (nullable = true)\n",
      " |-- weather_city_id: integer (nullable = true)\n",
      " |-- weather_city_name: string (nullable = true)\n",
      " |-- temp: double (nullable = true)\n",
      " |-- pressure: integer (nullable = true)\n",
      " |-- humidity: integer (nullable = true)\n",
      " |-- wind_speed: integer (nullable = true)\n",
      " |-- wind_deg: integer (nullable = true)\n",
      " |-- rain_3h: double (nullable = true)\n",
      " |-- clouds_all: integer (nullable = true)\n",
      " |-- weather_id: integer (nullable = true)\n",
      " |-- weather_main: string (nullable = true)\n",
      " |-- weather_description: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+---+-------+----+----------+------+----------------+------------+--------------+-----------------+-----------+-------+-------+--------+----------+------------------+----------+------------+------------+-----------+-----------+---------------+-----------------+------+--------+--------+----------+--------+-------+----------+----------+------------+-------------------+\n",
      "|year|  month|day|weekday|hour|atm_status|atm_id|atm_manufacturer|atm_location|atm_streetname|atm_street_number|atm_zipcode|atm_lat|atm_lon|currency| card_type|transaction_amount|   service|message_code|message_text|weather_lat|weather_lon|weather_city_id|weather_city_name|  temp|pressure|humidity|wind_speed|wind_deg|rain_3h|clouds_all|weather_id|weather_main|weather_description|\n",
      "+----+-------+---+-------+----+----------+------+----------------+------------+--------------+-----------------+-----------+-------+-------+--------+----------+------------------+----------+------------+------------+-----------+-----------+---------------+-----------------+------+--------+--------+----------+--------+-------+----------+----------+------------+-------------------+\n",
      "|2017|January|  1| Sunday|   0|    Active|     1|             NCR|  NÃƒÂ¦stved|   Farimagsvej|                8|       4700| 55.233| 11.763|     DKK|MasterCard|              5643|Withdrawal|        null|        null|      55.23|     11.761|        2616038|         Naestved|281.15|    1014|      87|         7|     260|  0.215|        92|       500|        Rain|         light rain|\n",
      "+----+-------+---+-------+----+----------+------+----------------+------------+--------------+-----------------+-----------+-------+-------+--------+----------+------------------+----------+------------+------------+-----------+-----------+---------------+-----------------+------+--------+--------+----------+--------+-------+----------+----------+------------+-------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['year',\n",
       " 'month',\n",
       " 'day',\n",
       " 'weekday',\n",
       " 'hour',\n",
       " 'atm_status',\n",
       " 'atm_id',\n",
       " 'atm_manufacturer',\n",
       " 'atm_location',\n",
       " 'atm_streetname',\n",
       " 'atm_street_number',\n",
       " 'atm_zipcode',\n",
       " 'atm_lat',\n",
       " 'atm_lon',\n",
       " 'currency',\n",
       " 'card_type',\n",
       " 'transaction_amount',\n",
       " 'service',\n",
       " 'message_code',\n",
       " 'message_text',\n",
       " 'weather_lat',\n",
       " 'weather_lon',\n",
       " 'weather_city_id',\n",
       " 'weather_city_name',\n",
       " 'temp',\n",
       " 'pressure',\n",
       " 'humidity',\n",
       " 'wind_speed',\n",
       " 'wind_deg',\n",
       " 'rain_3h',\n",
       " 'clouds_all',\n",
       " 'weather_id',\n",
       " 'weather_main',\n",
       " 'weather_description']"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating the Dimension and Fact tables"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating a dataframe for Location Dimension according to Target Dimension Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating a temporary df and selecting required columns & making sure records are distinct\n",
    "location = df.select('atm_location', 'atm_streetname', 'atm_street_number', 'atm_zipcode', 'atm_lat', 'atm_lon').distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+----------------+-----------------+-----------+-------+-------+-----------+\n",
      "|     atm_location|  atm_streetname|atm_street_number|atm_zipcode|atm_lat|atm_lon|location_id|\n",
      "+-----------------+----------------+-----------------+-----------+-------+-------+-----------+\n",
      "|            Vadum|  Ellehammersvej|               43|       9430| 57.118|  9.861|          0|\n",
      "|         Slagelse| Mariendals Alle|               29|       4200| 55.398| 11.342|          1|\n",
      "|       Fredericia|SjÃƒÂ¦llandsgade|               33|       7000| 55.564|  9.757|          2|\n",
      "|          Kolding|        Vejlevej|              135|       6000| 55.505|  9.457|          3|\n",
      "|HÃƒÂ¸rning Hallen|        Toftevej|               53|       8362| 56.091| 10.033|          4|\n",
      "+-----------------+----------------+-----------------+-----------+-------+-------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# creating the primary key column\n",
    "df_temp = location.rdd.zipWithIndex().toDF()\n",
    "dim_location = df_temp.select(col(\"_1.*\"),col(\"_2\").alias('location_id'))\n",
    "dim_location.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# renaming the colums as per requirement\n",
    "DIM_LOCATION = dim_location.withColumnRenamed('atm_location','location').withColumnRenamed('atm_streetname','streetname').withColumnRenamed('atm_street_number','street_number').withColumnRenamed('atm_zipcode','zipcode').withColumnRenamed('atm_lat','lat').withColumnRenamed('atm_lon','lon')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# rearranging the columns according to the target model\n",
    "DIM_LOCATION = DIM_LOCATION.select('location_id', 'location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['location_id',\n",
       " 'location',\n",
       " 'streetname',\n",
       " 'street_number',\n",
       " 'zipcode',\n",
       " 'lat',\n",
       " 'lon']"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "DIM_LOCATION.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "109"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "DIM_LOCATION.select('*').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "--------------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating a dataframe for ATM Dimension according to Target Dimension Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating a temporary df and selecting required columns\n",
    "atm = df.select('atm_id', 'atm_manufacturer', 'atm_lat', 'atm_lon')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# renaming the column atm_id to atm_number as per requirement\n",
    "atm = atm.withColumnRenamed('atm_id', 'atm_number')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# joining the dim_location and atm dataframes\n",
    "atm = atm.join(dim_location, on = ['atm_lat', 'atm_lon'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['atm_lat',\n",
       " 'atm_lon',\n",
       " 'atm_number',\n",
       " 'atm_manufacturer',\n",
       " 'atm_location',\n",
       " 'atm_streetname',\n",
       " 'atm_street_number',\n",
       " 'atm_zipcode',\n",
       " 'location_id']"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# checking to columns in the joined df\n",
    "atm.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# selecting the required columns and making sure records are distinct\n",
    "atm = atm.select('atm_number', 'atm_manufacturer', 'location_id').distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# renaming the colums as per requirement\n",
    "atm = atm.withColumnRenamed('location_id', 'atm_location_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['atm_number', 'atm_manufacturer', 'atm_location_id']"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# viewing changes in columns\n",
    "atm.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------------+---------------+------+\n",
      "|atm_number|atm_manufacturer|atm_location_id|atm_id|\n",
      "+----------+----------------+---------------+------+\n",
      "|        93|             NCR|            100|     0|\n",
      "|        68|             NCR|             80|     1|\n",
      "|        99|             NCR|             80|     2|\n",
      "|       106|             NCR|             53|     3|\n",
      "|        59| Diebold Nixdorf|             89|     4|\n",
      "+----------+----------------+---------------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# creating the primary key column\n",
    "df_temp = atm.rdd.zipWithIndex().toDF()\n",
    "dim_atm = df_temp.select(col(\"_1.*\"),col(\"_2\").alias('atm_id'))\n",
    "dim_atm.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "# rearranging the columns according to the target model\n",
    "DIM_ATM = dim_atm.select('atm_id', 'atm_number', 'atm_manufacturer', 'atm_location_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['atm_id', 'atm_number', 'atm_manufacturer', 'atm_location_id']"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "DIM_ATM.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "156"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "DIM_ATM.select('*').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "------------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating a dataframe for Date Dimension according to Target Dimension Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating a temporary df and selecting required columns\n",
    "date = df.select('year', 'month', 'day', 'hour', 'weekday')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "date = date.withColumn('full_date', concat_ws('-', date.year, date.month, date.day))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "date = date.withColumn('full_time', concat_ws(':', date.hour, lit('00'), lit('00')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "date = date.withColumn('full_date_time', concat_ws(' ', date.full_date, date.full_time))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "pattern = 'yyyy-MMM-dd HH:mm:ss'\n",
    "date = date.withColumn('full_date_time', unix_timestamp(date.full_date_time, pattern).cast('timestamp'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+---+----+-------+--------------+---------+-------------------+\n",
      "|year|month  |day|hour|weekday|full_date     |full_time|full_date_time     |\n",
      "+----+-------+---+----+-------+--------------+---------+-------------------+\n",
      "|2017|January|1  |0   |Sunday |2017-January-1|0:00:00  |2017-01-01 00:00:00|\n",
      "|2017|January|1  |0   |Sunday |2017-January-1|0:00:00  |2017-01-01 00:00:00|\n",
      "|2017|January|1  |0   |Sunday |2017-January-1|0:00:00  |2017-01-01 00:00:00|\n",
      "|2017|January|1  |0   |Sunday |2017-January-1|0:00:00  |2017-01-01 00:00:00|\n",
      "|2017|January|1  |0   |Sunday |2017-January-1|0:00:00  |2017-01-01 00:00:00|\n",
      "+----+-------+---+----+-------+--------------+---------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "date.show(5, truncate = False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "# selecting the required columns and making sure records are distinct\n",
    "date = date.select('year', 'month', 'day', 'hour', 'weekday', 'full_date_time').distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+---+----+---------+-------------------+-------+\n",
      "|year|  month|day|hour|  weekday|     full_date_time|date_id|\n",
      "+----+-------+---+----+---------+-------------------+-------+\n",
      "|2017|January|  1|  12|   Sunday|2017-01-01 12:00:00|      0|\n",
      "|2017|January|  6|   8|   Friday|2017-01-06 08:00:00|      1|\n",
      "|2017|January|  8|  16|   Sunday|2017-01-08 16:00:00|      2|\n",
      "|2017|January| 16|  18|   Monday|2017-01-16 18:00:00|      3|\n",
      "|2017|January| 18|   7|Wednesday|2017-01-18 07:00:00|      4|\n",
      "+----+-------+---+----+---------+-------------------+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# creating the primary key column\n",
    "df_temp = date.rdd.zipWithIndex().toDF()\n",
    "DIM_DATE = df_temp.select(col(\"_1.*\"),col(\"_2\").alias('date_id'))\n",
    "DIM_DATE.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "# rearranging the columns according to the target model\n",
    "DIM_DATE = DIM_DATE.select('date_id', 'full_date_time', 'year', 'month', 'day', 'hour', 'weekday')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['date_id', 'full_date_time', 'year', 'month', 'day', 'hour', 'weekday']"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "DIM_DATE.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8685"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "DIM_DATE.select('*').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "--------------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating a dataframe for Card Type Dimension according to Target Dimension Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating a temporary df and selecting required columns and making sure records are distinct\n",
    "card_type = df.select('card_type').distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+------------+\n",
      "|         card_type|card_type_id|\n",
      "+------------------+------------+\n",
      "|   Dankort - on-us|           0|\n",
      "|            CIRRUS|           1|\n",
      "|       HÃƒÂ¦vekort|           2|\n",
      "|              VISA|           3|\n",
      "|Mastercard - on-us|           4|\n",
      "+------------------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# creating the primary key column\n",
    "df_temp = card_type.rdd.zipWithIndex().toDF()\n",
    "DIM_CARD_TYPE = df_temp.select(col(\"_1.*\"),col(\"_2\").alias('card_type_id'))\n",
    "DIM_CARD_TYPE.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "# rearranging the columns according to the target model\n",
    "DIM_CARD_TYPE = DIM_CARD_TYPE.select('card_type_id', 'card_type')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['card_type_id', 'card_type']"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "DIM_CARD_TYPE.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "12"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "DIM_CARD_TYPE.select('*').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Creating the Transaction Fact Table according to Target Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stage 1 of FACT_ATM_TRANS Table -> joining original dataframe with DIM_LOC"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "# renaming the colums as per requirement\n",
    "fact_loc = df.withColumnRenamed('atm_location','location').withColumnRenamed('atm_streetname','streetname').withColumnRenamed('atm_street_number','street_number').withColumnRenamed('atm_zipcode','zipcode').withColumnRenamed('atm_lat','lat').withColumnRenamed('atm_lon','lon')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "# joining the dfs\n",
    "fact_loc = fact_loc.join(DIM_LOCATION, on = ['location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['location',\n",
       " 'streetname',\n",
       " 'street_number',\n",
       " 'zipcode',\n",
       " 'lat',\n",
       " 'lon',\n",
       " 'year',\n",
       " 'month',\n",
       " 'day',\n",
       " 'weekday',\n",
       " 'hour',\n",
       " 'atm_status',\n",
       " 'atm_id',\n",
       " 'atm_manufacturer',\n",
       " 'currency',\n",
       " 'card_type',\n",
       " 'transaction_amount',\n",
       " 'service',\n",
       " 'message_code',\n",
       " 'message_text',\n",
       " 'weather_lat',\n",
       " 'weather_lon',\n",
       " 'weather_city_id',\n",
       " 'weather_city_name',\n",
       " 'temp',\n",
       " 'pressure',\n",
       " 'humidity',\n",
       " 'wind_speed',\n",
       " 'wind_deg',\n",
       " 'rain_3h',\n",
       " 'clouds_all',\n",
       " 'weather_id',\n",
       " 'weather_main',\n",
       " 'weather_description',\n",
       " 'location_id']"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# viewing the columns\n",
    "fact_loc.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2468572"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Validating the count of the df at the end of Stage 1\n",
    "fact_loc.select('*').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stage 2 of FACT_ATM_TRANS Table -> joining the dataframe with DIM_ATM"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "# renaming the colums as per requirement\n",
    "fact_loc = fact_loc.withColumnRenamed('atm_id', 'atm_number').withColumnRenamed('location_id', 'atm_location_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "# joining the dfs\n",
    "fact_atm = fact_loc.join(DIM_ATM, on = ['atm_number', 'atm_manufacturer', 'atm_location_id'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "# performing necessary transformations, same as done to atm table\n",
    "fact_atm = fact_atm.withColumnRenamed('atm_location_id', 'weather_loc_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2468572"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Validating the count of the df at the end of Stage 2\n",
    "fact_atm.select('*').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stage 3 of FACT_ATM_TRANS Table -> joining the dataframe with DIM_DATE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "# joining the dfs\n",
    "fact_date = fact_atm.join(DIM_DATE, on = ['year', 'month', 'day', 'hour', 'weekday'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2468572"
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Validating the count of the df at the end of Stage 3\n",
    "fact_date.select('*').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stage 4 of FACT_ATM_TRANS Table -> joining the dataframe with DIM_CARD_TYPE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "# joining the dfs\n",
    "fact_atm_trans = fact_date.join(DIM_CARD_TYPE, on = ['card_type'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2468572"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Validating the count of the df at the end of Stage 4\n",
    "fact_atm_trans.select('*').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+----+-------+---+----+-------+----------+----------------+--------------+---------+--------------------+-------------+-------+------+----+----------+--------+------------------+----------+------------+------------+-----------+-----------+---------------+-----------------+-------+--------+--------+----------+--------+-------+----------+----------+------------+-------------------+------+-------+-------------------+------------+--------+\n",
      "|      card_type|year|  month|day|hour|weekday|atm_number|atm_manufacturer|weather_loc_id| location|          streetname|street_number|zipcode|   lat| lon|atm_status|currency|transaction_amount|   service|message_code|message_text|weather_lat|weather_lon|weather_city_id|weather_city_name|   temp|pressure|humidity|wind_speed|wind_deg|rain_3h|clouds_all|weather_id|weather_main|weather_description|atm_id|date_id|     full_date_time|card_type_id|trans_id|\n",
      "+---------------+----+-------+---+----+-------+----------+----------------+--------------+---------+--------------------+-------------+-------+------+----+----------+--------+------------------+----------+------------+------------+-----------+-----------+---------------+-----------------+-------+--------+--------+----------+--------+-------+----------+----------+------------+-------------------+------+-------+-------------------+------------+--------+\n",
      "|Dankort - on-us|2017|January|  1|  12| Sunday|        11|             NCR|            76|Sauersvej|Fridtjof Nansens Vej|            2|   9210|57.023|9.94|    Active|     DKK|              1621|Withdrawal|        null|        null|     57.048|      9.935|        2616235|   NÃƒÂ¸rresundby|278.521|    1017|      91|         6|     305|    0.0|        68|       803|      Clouds|      broken clouds|    23|      0|2017-01-01 12:00:00|           0|       1|\n",
      "+---------------+----+-------+---+----+-------+----------+----------------+--------------+---------+--------------------+-------------+-------+------+----+----------+--------+------------------+----------+------------+------------+-----------+-----------+---------------+-----------------+-------+--------+--------+----------+--------+-------+----------+----------+------------+-------------------+------+-------+-------------------+------------+--------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# creating primary key of fact table and viewing 1st record of the table\n",
    "from pyspark.sql.window import Window\n",
    "\n",
    "w = Window().orderBy('date_id')\n",
    "FACT_ATM_TRANS = fact_atm_trans.withColumn(\"trans_id\", row_number().over(w))\n",
    "FACT_ATM_TRANS.show(1, True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['card_type',\n",
       " 'year',\n",
       " 'month',\n",
       " 'day',\n",
       " 'hour',\n",
       " 'weekday',\n",
       " 'atm_number',\n",
       " 'atm_manufacturer',\n",
       " 'weather_loc_id',\n",
       " 'location',\n",
       " 'streetname',\n",
       " 'street_number',\n",
       " 'zipcode',\n",
       " 'lat',\n",
       " 'lon',\n",
       " 'atm_status',\n",
       " 'currency',\n",
       " 'transaction_amount',\n",
       " 'service',\n",
       " 'message_code',\n",
       " 'message_text',\n",
       " 'weather_lat',\n",
       " 'weather_lon',\n",
       " 'weather_city_id',\n",
       " 'weather_city_name',\n",
       " 'temp',\n",
       " 'pressure',\n",
       " 'humidity',\n",
       " 'wind_speed',\n",
       " 'wind_deg',\n",
       " 'rain_3h',\n",
       " 'clouds_all',\n",
       " 'weather_id',\n",
       " 'weather_main',\n",
       " 'weather_description',\n",
       " 'atm_id',\n",
       " 'date_id',\n",
       " 'full_date_time',\n",
       " 'card_type_id',\n",
       " 'trans_id']"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# viewing the list of columns\n",
    "FACT_ATM_TRANS.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "# selecting and arranging only the required columns according to the target model\n",
    "FACT_ATM_TRANS = FACT_ATM_TRANS.select('trans_id', 'atm_id', 'weather_loc_id', 'date_id', 'card_type_id', \n",
    "'atm_status', 'currency', 'service', 'transaction_amount', 'message_code', 'message_text', 'rain_3h', \n",
    "'clouds_all', 'weather_id', 'weather_main', 'weather_description')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['trans_id',\n",
       " 'atm_id',\n",
       " 'weather_loc_id',\n",
       " 'date_id',\n",
       " 'card_type_id',\n",
       " 'atm_status',\n",
       " 'currency',\n",
       " 'service',\n",
       " 'transaction_amount',\n",
       " 'message_code',\n",
       " 'message_text',\n",
       " 'rain_3h',\n",
       " 'clouds_all',\n",
       " 'weather_id',\n",
       " 'weather_main',\n",
       " 'weather_description']"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "FACT_ATM_TRANS.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2468572"
      ]
     },
     "execution_count": 66,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "FACT_ATM_TRANS.select('*').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "--------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Writing the PySpark Dataframes to AWS S3 Storage in csv format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [],
   "source": [
    "# writing data from pyspark df 'dim_location' in csv format to dim_location folder in S3 bucket 'etlprojectbypriya'\n",
    "DIM_LOCATION.coalesce(1).write.format('csv').option('header','false').save('s3a://etlprojectbysimran/dim_location', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [],
   "source": [
    "# writing data from pyspark df 'dim_atm' in csv format to dim_atm folder in S3 bucket 'etlprojectbypriya'\n",
    "DIM_ATM.coalesce(1).write.format('csv').option('header','false').save('s3a://etlprojectbysimran/dim_atm', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "# writing data from pyspark df 'dim_data' in csv format to dim_data folder in S3 bucket 'etlprojectbypriya'\n",
    "DIM_DATE.coalesce(1).write.format('csv').option('header','false').save('s3a://etlprojectbysimran/dim_date', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [],
   "source": [
    "# writing data from pyspark df 'dim_card_type' in csv format to dim_card_type folder in S3 bucket 'etlprojectbypriya'\n",
    "DIM_CARD_TYPE.coalesce(1).write.format('csv').option('header','false').save('s3a://etlprojectbysimran/dim_card_type', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "# writing data from pyspark df 'fact_atm_trans' in csv format to fact_atm_trans folder in S3 bucket 'etlprojectbypriya'\n",
    "FACT_ATM_TRANS.coalesce(1).write.format('csv').option('header','false').save('s3a://etlprojectbysimran/fact_atm_trans', mode='overwrite')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "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.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}