etl_3NF_to_factdimtables.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ETL the data from 3NF tables to Facts & Dimension Tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The sql extension is already loaded. To reload it, use:\n",
      "  %reload_ext sql\n",
      "postgresql://*****:******@127.0.0.1:5432/pagila\n"
     ]
    }
   ],
   "source": [
    "%load_ext sql\n",
    "\n",
    "DB_ENDPOINT = \"127.0.0.1\"\n",
    "DB = 'pagila'\n",
    "DB_USER = '*****'\n",
    "DB_PASSWORD = '******'\n",
    "DB_PORT = '5432'\n",
    "\n",
    "# postgresql://username:password@host:port/database\n",
    "conn_string = \"postgresql://{}:{}@{}:{}/{}\" \\\n",
    "                        .format(DB_USER, DB_PASSWORD, DB_ENDPOINT, DB_PORT, DB)\n",
    "\n",
    "print(conn_string)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%sql $conn_string"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `Extract` data from the normalized database, `transform` it, and `load` it into the new tables. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * postgresql://postgres:***@127.0.0.1:5432/pagila\n",
      "40 rows affected.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql\n",
    "INSERT INTO dimDate (date_key, date, year, quarter, month, day, week, is_weekend)\n",
    "SELECT DISTINCT(TO_CHAR(payment_date :: DATE, 'yyyyMMDD')::integer) AS date_key,\n",
    "       date(payment_date)                                           AS date,\n",
    "       EXTRACT(year FROM payment_date)                              AS year,\n",
    "       EXTRACT(quarter FROM payment_date)                           AS quarter,\n",
    "       EXTRACT(month FROM payment_date)                             AS month,\n",
    "       EXTRACT(day FROM payment_date)                               AS day,\n",
    "       EXTRACT(week FROM payment_date)                              AS week,\n",
    "       CASE WHEN EXTRACT(ISODOW FROM payment_date) IN (6, 7) THEN true ELSE false END AS is_weekend\n",
    "FROM payment;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * postgresql://postgres:***@127.0.0.1:5432/pagila\n",
      "599 rows affected.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql\n",
    "INSERT INTO dimCustomer (customer_key, customer_id, first_name, last_name, email, address, \n",
    "                         address2, district, city, country, postal_code, phone, active, \n",
    "                         create_date, start_date, end_date)\n",
    "SELECT c.customer_id AS customer_key,\n",
    "       c.customer_id AS customer_id,\n",
    "       c.first_name  AS first_name,\n",
    "       c.last_name   AS last_name,\n",
    "       c.email       AS email,\n",
    "       a.address     AS address,\n",
    "       a.address2    AS address2,\n",
    "       a.district    AS district,\n",
    "       ci.city       AS city,\n",
    "       co.country    AS country,\n",
    "       a.postal_code AS postal_code,\n",
    "       a.phone       AS phone,\n",
    "       c.active      AS active,\n",
    "       c.create_date AS creat_date,\n",
    "       now()         AS start_date,\n",
    "       now()         AS end_date\n",
    "FROM customer c\n",
    "JOIN address a  ON (c.address_id = a.address_id)\n",
    "JOIN city ci    ON (a.city_id = ci.city_id)\n",
    "JOIN country co ON (ci.country_id = co.country_id);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * postgresql://postgres:***@127.0.0.1:5432/pagila\n",
      "1000 rows affected.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql\n",
    "INSERT INTO dimMovie(movie_key, film_id, title, description, release_year, \n",
    "                     language, original_language, rental_duration, length, rating, \n",
    "                     special_features)\n",
    "SELECT \n",
    "        f.film_id           AS movie_key,\n",
    "        f.film_id           AS film_id,\n",
    "        f.title             AS title,\n",
    "        f.description       AS description,\n",
    "        f.release_year      AS release_year,\n",
    "        l.name              AS language,  \n",
    "        orig_lang.name      AS original_language,\n",
    "        f.rental_duration   AS rental_duration,\n",
    "        f.length            AS length,\n",
    "        f.rating            AS rating,\n",
    "        f.special_features  AS special_features\n",
    "FROM film f\n",
    "JOIN language l              ON (f.language_id=l.language_id)\n",
    "LEFT JOIN language orig_lang ON (f.original_language_id = orig_lang.language_id);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * postgresql://postgres:***@127.0.0.1:5432/pagila\n",
      "2 rows affected.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql\n",
    "INSERT INTO dimStore(store_key, store_id, address, address2, district, city, country, \n",
    "                     postal_code, manager_first_name, manager_last_name, \n",
    "                     start_date, end_date)\n",
    "SELECT s.store_id  AS store_key,\n",
    "       s.store_id  AS store_id,\n",
    "       a.address   AS address,\n",
    "       a.address2  AS address2,\n",
    "       a.district  AS district,\n",
    "       ci.city     AS city,\n",
    "       co.country  AS country,\n",
    "       a.postal_code AS postal_code,\n",
    "       st.first_name AS manager_first_name,\n",
    "       st.last_name  AS manager_last_name,\n",
    "       now()         AS start_date,\n",
    "       now()         AS end_date\n",
    "FROM store s\n",
    "JOIN staff st   ON (s.manager_staff_id = st.staff_id)\n",
    "JOIN address a  ON (s.address_id = a.address_id)\n",
    "JOIN city ci    ON (a.city_id = ci.city_id)\n",
    "JOIN country co ON (ci.country_id = co.country_id);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * postgresql://postgres:***@127.0.0.1:5432/pagila\n",
      "16049 rows affected.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql\n",
    "INSERT INTO factSales(date_key, customer_key, movie_key, store_key, sales_amount)\n",
    "SELECT TO_CHAR(p.payment_date :: DATE, 'yyyyMMDD')::integer AS date_key,\n",
    "       p.customer_id  AS customer_key,\n",
    "       i.film_id    AS movie_key,\n",
    "       i.store_id   AS store_key,\n",
    "       p.amount     AS sales_amount\n",
    "FROM payment p\n",
    "JOIN rental r    ON (p.rental_id = r.rental_id)\n",
    "JOIN inventory i ON (r.inventory_id = i.inventory_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.8.3"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}