SparkNordBank_SparkETLCode.py.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "21b9a85a",
   "metadata": {},
   "source": [
    "# SPAR NORD ETL PROJECT (BY ANKIT ABHISHEK MANIKANDAN BHARAT KUMAR)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "1d77f792",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>User</th><th>Current session?</th></tr><tr><td>4</td><td>application_1714904106262_0006</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-88-57.ec2.internal:20888/proxy/application_1714904106262_0006/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-88-57.ec2.internal:8042/node/containerlogs/container_1714904106262_0006_01_000001/livy\">Link</a></td><td>None</td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<pyspark.sql.session.SparkSession object at 0x7f8d69dded50>"
     ]
    }
   ],
   "source": [
    "#SPARK SESSION CONNECTIVITY CONFIRMATION\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3955a820",
   "metadata": {},
   "source": [
    "## 1. READING DATA TO SPARK"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1fa01fc",
   "metadata": {},
   "source": [
    "### 1.1 Reading data from HDFS to Spark using default schema for getting confirmation of data is loaded or not "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "517a143c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "spfr =spark.read.csv(\"/user/root/SRC_ATM_TRANS/part-m-00000\", header = False, inferSchema = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "20fa58ef",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "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"
     ]
    }
   ],
   "source": [
    "spfr.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "1f70251b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: integer (nullable = true)\n",
      " |-- _c1: string (nullable = true)\n",
      " |-- _c2: integer (nullable = true)\n",
      " |-- _c3: string (nullable = true)\n",
      " |-- _c4: integer (nullable = true)\n",
      " |-- _c5: string (nullable = true)\n",
      " |-- _c6: integer (nullable = true)\n",
      " |-- _c7: string (nullable = true)\n",
      " |-- _c8: string (nullable = true)\n",
      " |-- _c9: string (nullable = true)\n",
      " |-- _c10: integer (nullable = true)\n",
      " |-- _c11: integer (nullable = true)\n",
      " |-- _c12: double (nullable = true)\n",
      " |-- _c13: double (nullable = true)\n",
      " |-- _c14: string (nullable = true)\n",
      " |-- _c15: string (nullable = true)\n",
      " |-- _c16: integer (nullable = true)\n",
      " |-- _c17: string (nullable = true)\n",
      " |-- _c18: integer (nullable = true)\n",
      " |-- _c19: string (nullable = true)\n",
      " |-- _c20: double (nullable = true)\n",
      " |-- _c21: double (nullable = true)\n",
      " |-- _c22: integer (nullable = true)\n",
      " |-- _c23: string (nullable = true)\n",
      " |-- _c24: double (nullable = true)\n",
      " |-- _c25: integer (nullable = true)\n",
      " |-- _c26: integer (nullable = true)\n",
      " |-- _c27: integer (nullable = true)\n",
      " |-- _c28: integer (nullable = true)\n",
      " |-- _c29: double (nullable = true)\n",
      " |-- _c30: integer (nullable = true)\n",
      " |-- _c31: integer (nullable = true)\n",
      " |-- _c32: string (nullable = true)\n",
      " |-- _c33: string (nullable = true)"
     ]
    }
   ],
   "source": [
    "spfr.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2cf54d73",
   "metadata": {},
   "source": [
    "### 1.2 Converting default input schema to custom input schema and reading data "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "29fc8f34",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType, DoubleType, LongType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "76d9285e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "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,
   "id": "0af10e73",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "spfr = spark.read.csv(\"/user/root/SRC_ATM_TRANS/part-m-00000\", header = False, schema = Schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "2aa19d5e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2468572"
     ]
    }
   ],
   "source": [
    "spfr.select('*').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "e83a7b2d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "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)"
     ]
    }
   ],
   "source": [
    "spfr.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "455e67b0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "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"
     ]
    }
   ],
   "source": [
    "spfr.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "4f24d236",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['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']"
     ]
    }
   ],
   "source": [
    "spfr.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e59a772e",
   "metadata": {},
   "source": [
    "## 2. CREATION OF DIMENSION TABLES"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "332179bc",
   "metadata": {},
   "source": [
    "### 2.1 Location Dimension table creation according to target dimension model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "5315b6c6",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "5c66c40c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "109"
     ]
    }
   ],
   "source": [
    "# creating a temporary df and selecting required columns & making sure records are distinct\n",
    "locate = spfr.select('atm_location', 'atm_streetname', 'atm_street_number', 'atm_zipcode', 'atm_lat', 'atm_lon').distinct()\n",
    "locate.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "a15a4f02",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+-------------------+-----------------+-----------+-------+-------+-----------+\n",
      "|    atm_location|     atm_streetname|atm_street_number|atm_zipcode|atm_lat|atm_lon|location_id|\n",
      "+----------------+-------------------+-----------------+-----------+-------+-------+-----------+\n",
      "|         Kolding|           Vejlevej|              135|       6000| 55.505|  9.457|          0|\n",
      "|  Skelagervej 15|        Skelagervej|               15|       9000| 57.023|  9.891|          1|\n",
      "|Intern HolbÃƒÂ¦k|        Slotsvolden|                7|       4300| 55.718| 11.704|          2|\n",
      "|          Odense|       FÃƒÂ¦lledvej|                3|       5000| 55.394|  10.37|          3|\n",
      "|           Ikast|RÃƒÂ¥dhusstrÃƒÂ¦det|               12|       7430| 56.139|  9.154|          4|\n",
      "+----------------+-------------------+-----------------+-----------+-------+-------+-----------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "# Creating LOCATION_ID as primary key in the dataframe \n",
    "spfr_temp = locate.rdd.zipWithIndex().toDF()\n",
    "dim_locate = spfr_temp.select(col(\"_1.*\"),col(\"_2\").alias('location_id'))\n",
    "dim_locate.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "234629ae",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# renaming the colums as per requirement\n",
    "dim_locate = dim_locate.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,
   "id": "0c8eb306",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# rearranging the columns according to the target model\n",
    "dim_locate = dim_locate.select('location_id', 'location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "d3167041",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['location_id', 'location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon']"
     ]
    }
   ],
   "source": [
    "# Columns of the location dimension table\n",
    "dim_locate.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "4d721f63",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------------+-------------------+-------------+-------+------+------+\n",
      "|location_id|        location|         streetname|street_number|zipcode|   lat|   lon|\n",
      "+-----------+----------------+-------------------+-------------+-------+------+------+\n",
      "|          0|         Kolding|           Vejlevej|          135|   6000|55.505| 9.457|\n",
      "|          1|  Skelagervej 15|        Skelagervej|           15|   9000|57.023| 9.891|\n",
      "|          2|Intern HolbÃƒÂ¦k|        Slotsvolden|            7|   4300|55.718|11.704|\n",
      "|          3|          Odense|       FÃƒÂ¦lledvej|            3|   5000|55.394| 10.37|\n",
      "|          4|           Ikast|RÃƒÂ¥dhusstrÃƒÂ¦det|           12|   7430|56.139| 9.154|\n",
      "+-----------+----------------+-------------------+-------------+-------+------+------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "# fetching first 5 row values\n",
    "dim_locate.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "6ec1be77",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "109"
     ]
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "dim_locate.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d19f9db1",
   "metadata": {},
   "source": [
    "### 2.2 ATM Dimension table creation according to target dimension model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "15d7fb5d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# creating a temporary atm df and selecting required columns\n",
    "atm = spfr.select('atm_id', 'atm_manufacturer', 'atm_lat', 'atm_lon')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "423d9e25",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# renaming the column atm_id to atm_number as per requirement\n",
    "atm = atm.withColumnRenamed('atm_id', 'atm_number').withColumnRenamed('atm_lat','lat').withColumnRenamed('atm_lon','lon')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "dee2f921",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Join dim_locate through left join on atm \n",
    "atm = atm.join(dim_locate, on = ['lat', 'lon'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "925ed1ee",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['lat', 'lon', 'atm_number', 'atm_manufacturer', 'location_id', 'location', 'streetname', 'street_number', 'zipcode']"
     ]
    }
   ],
   "source": [
    "# Columns after doing left join \n",
    "atm.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "efb20975",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#Rearranging according to requirement\n",
    "atm = atm.select('atm_number', 'atm_manufacturer', 'location_id').distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "08ebd35b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#Renaming according to requirement\n",
    "atm = atm.withColumnRenamed('location_id', 'atm_location_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "d4fcf9d6",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['atm_number', 'atm_manufacturer', 'atm_location_id']"
     ]
    }
   ],
   "source": [
    "atm.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "9a5e6fd2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------------+---------------+------+\n",
      "|atm_number|atm_manufacturer|atm_location_id|atm_id|\n",
      "+----------+----------------+---------------+------+\n",
      "|       100|             NCR|             70|     0|\n",
      "|        60|             NCR|             68|     1|\n",
      "|        45|             NCR|             66|     2|\n",
      "|        54|             NCR|              9|     3|\n",
      "|        53|             NCR|             27|     4|\n",
      "+----------+----------------+---------------+------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "# Creating atm_id as primary key in the dataframe \n",
    "spfr_temp = atm.rdd.zipWithIndex().toDF()\n",
    "dim_atm = spfr_temp.select(col(\"_1.*\"),col(\"_2\").alias('atm_id'))\n",
    "dim_atm.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "5430f3b5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "156"
     ]
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "dim_atm.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b050666a",
   "metadata": {},
   "source": [
    "### 2.3 Date Dimension table creation according to target dimension model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "6a29ddab",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "date = spfr.select('year', 'month', 'day', 'hour', 'weekday')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "a6d0d931",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "date = date.withColumn('full_date_time', to_timestamp(concat(date.year, lit('-'), date.month, lit('-'),date.day, lit(' '), date.hour), 'yyyy-MMMM-d H'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "fe8939ea",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2468572"
     ]
    }
   ],
   "source": [
    "date.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "2e8d3c73",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# selecting the required columns and making sure records are distinct\n",
    "date = date.select('full_date_time', 'year', 'month', 'day', 'hour', 'weekday').distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "82a5f6c5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----+--------+---+----+-------+-------+\n",
      "|     full_date_time|year|   month|day|hour|weekday|date_id|\n",
      "+-------------------+----+--------+---+----+-------+-------+\n",
      "|2017-01-06 08:00:00|2017| January|  6|   8| Friday|      0|\n",
      "|2017-01-16 18:00:00|2017| January| 16|  18| Monday|      1|\n",
      "|2017-01-29 08:00:00|2017| January| 29|   8| Sunday|      2|\n",
      "|2017-02-26 08:00:00|2017|February| 26|   8| Sunday|      3|\n",
      "|2017-03-14 06:00:00|2017|   March| 14|   6|Tuesday|      4|\n",
      "+-------------------+----+--------+---+----+-------+-------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "# Creating date_id as primary key in the dataframe\n",
    "spfr_temp = date.rdd.zipWithIndex().toDF()\n",
    "dim_date = spfr_temp.select(col(\"_1.*\"),col(\"_2\").alias('date_id'))\n",
    "dim_date.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "aa5a5d21",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "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": 35,
   "id": "dffeb091",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['date_id', 'full_date_time', 'year', 'month', 'day', 'hour', 'weekday']"
     ]
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "dim_date.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "ad4a734f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8685"
     ]
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "dim_date.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c08a54fa",
   "metadata": {},
   "source": [
    "### 2.4 Card Type Dimension table creation according to target dimension model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "e2b2e279",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# creating a temporary atmcard_type df and selecting required columns and making sure records are distinct\n",
    "atmcard_type =  spfr.select('card_type').distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "f5cc9d28",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------+\n",
      "|           card_type|card_type_id|\n",
      "+--------------------+------------+\n",
      "|Visa Dankort - on-us|           0|\n",
      "|  Mastercard - on-us|           1|\n",
      "|         HÃƒÂ¦vekort|           2|\n",
      "|            VisaPlus|           3|\n",
      "|     Dankort - on-us|           4|\n",
      "+--------------------+------------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "# Creating card_type_id as primary key in the dataframe\n",
    "spfr_temp = atmcard_type.rdd.zipWithIndex().toDF()\n",
    "atmcard_type = spfr_temp.select(col(\"_1.*\"),col(\"_2\").alias('card_type_id'))\n",
    "atmcard_type.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "8b3cff09",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# rearranging the columns according to the target model\n",
    "atmcard_type = atmcard_type.select('card_type_id', 'card_type')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "3d1e8159",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['card_type_id', 'card_type']"
     ]
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "atmcard_type.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "890129c5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "12"
     ]
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "atmcard_type.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1a50eeb",
   "metadata": {},
   "source": [
    "## 3. CREATION OF TRANSACTION FACT TABLES"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "672371c7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Stage 1 of FACT_ATM_TRANS Table -> joining original dataframe with DIM_LOC\n",
    "# renaming the colums as per requirement\n",
    "fact_locate = spfr.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": 43,
   "id": "2ac17dbd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# joining original dataframe with DIM_LOCATION\n",
    "fact_locate = fact_locate.join(dim_locate, on = ['location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "67e9468c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2468572"
     ]
    }
   ],
   "source": [
    "fact_locate.count()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "6d5608aa",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon', 'year', 'month', 'day', 'weekday', 'hour', 'atm_status', 'atm_id', 'atm_manufacturer', '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', 'location_id']"
     ]
    }
   ],
   "source": [
    "fact_locate.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "id": "1f26dbca",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "fact_locate = fact_locate.withColumnRenamed('atm_id', 'atm_number').withColumnRenamed('location_id','atm_location_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "853944da",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon', 'year', 'month', 'day', 'weekday', 'hour', 'atm_status', 'atm_number', 'atm_manufacturer', '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', 'atm_location_id']"
     ]
    }
   ],
   "source": [
    "fact_locate.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "eff00ded",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# joining the dataframe with DIM_ATM\n",
    "fact_atm = fact_locate.join(dim_atm, on = ['atm_number', 'atm_manufacturer', 'atm_location_id'], how = \"left\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "id": "470eede1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['atm_number', 'atm_manufacturer', 'atm_location_id', 'location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon', 'year', 'month', 'day', 'weekday', 'hour', 'atm_status', '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', 'atm_id']"
     ]
    }
   ],
   "source": [
    "fact_atm.columns\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "id": "d0401e88",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "fact_atm = fact_atm.withColumnRenamed('atm_location_id', 'weather_loc_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "id": "56c73a9d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# joining the dataframe with DIM_DATE\n",
    "fact_date = fact_atm.join(dim_date, on = ['year', 'month', 'day', 'hour', 'weekday'], how = \"left\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "id": "68f37e2f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['year', 'month', 'day', 'hour', 'weekday', 'atm_number', 'atm_manufacturer', 'weather_loc_id', 'location', 'streetname', 'street_number', 'zipcode', 'lat', 'lon', 'atm_status', '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', 'atm_id', 'date_id', 'full_date_time']"
     ]
    }
   ],
   "source": [
    "fact_date.columns\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "id": "981d2eae",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# joining the dataframe with DIM_CARD_TYPE\n",
    "fact_atmtransaction = fact_date.join(atmcard_type, on = ['card_type'], how = \"left\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "id": "93a2d96f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['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']"
     ]
    }
   ],
   "source": [
    "fact_atmtransaction.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "id": "6479f99a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql.window import Window\n",
    "fact_atmtransaction = fact_atmtransaction.withColumn(\"trans_id\", row_number().over(Window.orderBy('date_id')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "id": "268cb43e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['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']"
     ]
    }
   ],
   "source": [
    "fact_atmtransaction.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "id": "048c4d14",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# selecting and arranging only the required columns according to the target model\n",
    "fact_atmtransaction = fact_atmtransaction.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": 58,
   "id": "47e4be0d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['trans_id', 'atm_id', 'weather_loc_id', 'date_id', 'card_type_id', 'atm_status', 'currency', 'service', 'transaction_amount', 'message_code', 'message_text', 'rain_3h', 'clouds_all', 'weather_id', 'weather_main', 'weather_description']"
     ]
    }
   ],
   "source": [
    "# checking that all required columns are present and named correctly\n",
    "fact_atmtransaction.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "id": "235540e2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2468572"
     ]
    }
   ],
   "source": [
    "# validating the count of the dataframe\n",
    "fact_atmtransaction.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "040d20e3",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+--------------+-------+------------+----------+--------+----------+------------------+------------+------------+-------+----------+----------+------------+-------------------+\n",
      "|trans_id|atm_id|weather_loc_id|date_id|card_type_id|atm_status|currency|   service|transaction_amount|message_code|message_text|rain_3h|clouds_all|weather_id|weather_main|weather_description|\n",
      "+--------+------+--------------+-------+------------+----------+--------+----------+------------------+------------+------------+-------+----------+----------+------------+-------------------+\n",
      "|       1|   105|             1|      0|          10|  Inactive|     DKK|Withdrawal|              4447|        null|        null|    0.0|        12|       801|      Clouds|         few clouds|\n",
      "|       2|   134|            15|      0|           1|    Active|     DKK|Withdrawal|              6191|        null|        null|    0.0|        36|       802|      Clouds|   scattered clouds|\n",
      "|       3|   142|            95|      0|           0|    Active|     DKK|Withdrawal|              9483|        null|        null|    0.0|         8|       800|       Clear|       sky is clear|\n",
      "|       4|     3|             9|      0|           1|  Inactive|     DKK|Withdrawal|              8658|        null|        null|    0.0|        24|       801|      Clouds|         few clouds|\n",
      "|       5|    69|            54|      0|           4|    Active|     DKK|Withdrawal|               367|        null|        null|    0.0|        12|       801|      Clouds|         few clouds|\n",
      "+--------+------+--------------+-------+------------+----------+--------+----------+------------------+------------+------------+-------+----------+----------+------------+-------------------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "fact_atmtransaction.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "61698090",
   "metadata": {},
   "source": [
    "## 4. EXPORTING PYSPARK DATAFRAMES TO AWS S3 STORAGE IN CSV FORMAT"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "5604b0c4",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# writing data from pyspark df 'dim_locate' in csv format to dim_locate folder in S3 bucket 'sparnordetlproject'\n",
    "dim_locate.coalesce(1).write.format('csv').option('header','false').save('s3a://sparnordetlproject/dim_locate', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "id": "17d8b824",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# writing data from pyspark df 'dim_atm' in csv format to dim_atm folder in S3 bucket 'sparnordetlproject'\n",
    "dim_atm.coalesce(1).write.format('csv').option('header','false').save('s3a://sparnordetlproject/dim_atm', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "id": "8831db20",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# writing data from pyspark df 'dim_date' in csv format to dim_date folder in S3 bucket 'sparnordetlproject'\n",
    "dim_date.coalesce(1).write.format('csv').option('header','false').save('s3a://sparnordetlproject/dim_date', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "id": "edf6584e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# writing data from pyspark df 'atmcard_type' in csv format to atmcard_type folder in S3 bucket 'sparnordetlproject'\n",
    "atmcard_type.coalesce(1).write.format('csv').option('header','false').save('s3a://sparnordetlproject/atmcard_type', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "id": "ee12d12f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# writing data from pyspark df 'fact_atmtransaction' in csv format to fact_atmtransaction folder in S3 bucket 'sparnordetlproject'\n",
    "fact_atmtransaction.coalesce(1).write.format('csv').option('header','false').save('s3a://sparnordetlproject/fact_atmtransaction', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "59f183e0",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.10.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}