etl-pl-pgsql.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "id": "04918d3b-ea44-4916-8d03-8d9ff6dc0b7b",
   "metadata": {},
   "source": [
    "---\n",
    "title: \"ETL usando PL/pgSQL (Procedural Language/PostgreSQL)\"    \n",
    "format:    \n",
    "  pdf:    \n",
    "    documentclass: article    \n",
    "    keep-tex: true    \n",
    "    geometry: a4paper,landscape,margin=1cm \n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d80da7b7",
   "metadata": {},
   "source": [
    "# ETL usando PL/pgSQL (Procedural Language/PostgreSQL).\n",
    "\n",
    "---\n",
    "\n",
    "## Configuración Inicial\n",
    "\n",
    "Primero, configuraremos el entorno como antes, instalando PostgreSQL y creando las bases de datos `pagila` y `pagila_dw`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "556c66ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Instalar los paquetes necesarios\n",
    "!pip install ipython-sql prettytable==0.7.2 psycopg2-binary\n",
    "\n",
    "# Actualizar los paquetes e instalar PostgreSQL\n",
    "!apt-get update -y\n",
    "!apt-get install postgresql postgresql-contrib -y\n",
    "\n",
    "# Iniciar el servicio de PostgreSQL\n",
    "import os\n",
    "os.system('service postgresql start')\n",
    "\n",
    "# Cambiar la contraseña del usuario 'postgres'\n",
    "!sudo -u postgres psql -c \"ALTER USER postgres PASSWORD 'postgres';\"\n",
    "\n",
    "# Crear las bases de datos 'pagila' y 'pagila_dw'\n",
    "!sudo -u postgres createdb pagila\n",
    "!sudo -u postgres createdb pagila_dw\n",
    "\n",
    "# Descargar el script de creación del esquema de Pagila\n",
    "!wget https://raw.githubusercontent.com/devrimgunduz/pagila/master/pagila-schema.sql\n",
    "\n",
    "# Descargar el script de inserción de datos de Pagila\n",
    "!wget https://raw.githubusercontent.com/devrimgunduz/pagila/master/pagila-data.sql\n",
    "\n",
    "# Ejecutar el script de creación del esquema en 'pagila'\n",
    "!sudo -u postgres psql pagila < pagila-schema.sql\n",
    "\n",
    "# Ejecutar el script de inserción de datos en 'pagila'\n",
    "!sudo -u postgres psql pagila < pagila-data.sql"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31e18870",
   "metadata": {},
   "source": [
    "## Configuración de `postgres_fdw` y Esquemas\n",
    "\n",
    "Configuraremos `postgres_fdw` para acceder a los datos de `pagila` desde `pagila_dw` y crearemos los esquemas necesarios."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9b2d918a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import psycopg2\n",
    "\n",
    "# Conectar a la base de datos pagila_dw\n",
    "conn_dw = psycopg2.connect(\n",
    "    dbname=\"pagila_dw\",\n",
    "    user=\"postgres\",\n",
    "    password=\"postgres\",\n",
    "    host=\"localhost\",\n",
    "    port=\"5432\"\n",
    ")\n",
    "\n",
    "cursor_dw = conn_dw.cursor()\n",
    "\n",
    "# Crear el esquema 'dw' en pagila_dw\n",
    "cursor_dw.execute(\"CREATE SCHEMA IF NOT EXISTS dw;\")\n",
    "conn_dw.commit()\n",
    "\n",
    "# Instalar la extensión postgres_fdw\n",
    "cursor_dw.execute(\"CREATE EXTENSION IF NOT EXISTS postgres_fdw;\")\n",
    "conn_dw.commit()\n",
    "\n",
    "# Crear el servidor FDW para conectarse a pagila\n",
    "cursor_dw.execute(\"\"\"\n",
    "CREATE SERVER IF NOT EXISTS pagila_server\n",
    "FOREIGN DATA WRAPPER postgres_fdw\n",
    "OPTIONS (host 'localhost', dbname 'pagila', port '5432');\n",
    "\"\"\")\n",
    "conn_dw.commit()\n",
    "\n",
    "# Crear el mapeo de usuario\n",
    "cursor_dw.execute(\"\"\"\n",
    "CREATE USER MAPPING IF NOT EXISTS FOR postgres\n",
    "SERVER pagila_server\n",
    "OPTIONS (user 'postgres', password 'postgres');\n",
    "\"\"\")\n",
    "conn_dw.commit()\n",
    "\n",
    "# Importar el esquema público de pagila como esquema 'pagila' en pagila_dw\n",
    "cursor_dw.execute(\"\"\"\n",
    "DROP SCHEMA IF EXISTS pagila CASCADE;\n",
    "CREATE SCHEMA pagila;\n",
    "IMPORT FOREIGN SCHEMA public\n",
    "FROM SERVER pagila_server\n",
    "INTO pagila;\n",
    "\"\"\")\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "47dbe988",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## Creación de los Procedimientos Almacenados con Upsert\n",
    "\n",
    "### Consideraciones Previas\n",
    "\n",
    "Para utilizar la funcionalidad de upsert (`INSERT ... ON CONFLICT DO UPDATE`), necesitamos definir claves únicas o primarias en las tablas de destino que nos permitan identificar los registros para la operación de actualización. Añadiremos restricciones de clave primaria o única en los campos adecuados de las tablas de dimensiones y hechos.\n",
    "\n",
    "### Dimensión Película\n",
    "\n",
    "#### Creación de la Tabla con Clave Primaria\n",
    "\n",
    "Primero, creamos la tabla `dw.dim_pelicula` con una clave primaria en `film_id`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5c73a610",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_create_dim_pelicula_table = \"\"\"\n",
    "CREATE TABLE IF NOT EXISTS dw.dim_pelicula (\n",
    "    pelicula_key SERIAL PRIMARY KEY,\n",
    "    film_id INTEGER UNIQUE,\n",
    "    titulo TEXT,\n",
    "    descripcion TEXT,\n",
    "    anio_lanzamiento INTEGER,\n",
    "    idioma TEXT,\n",
    "    duracion_renta INTEGER,\n",
    "    tasa_renta NUMERIC,\n",
    "    duracion INTEGER,\n",
    "    costo_reemplazo NUMERIC,\n",
    "    clasificacion TEXT,\n",
    "    caracteristicas_especiales TEXT,\n",
    "    categoria TEXT,\n",
    "    fecha_ultima_actualizacion TIMESTAMP\n",
    ");\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_create_dim_pelicula_table)\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "540a0d73",
   "metadata": {},
   "source": [
    "#### Procedimiento Almacenado con Upsert"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cc39d04f",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_dim_pelicula_upsert = \"\"\"\n",
    "CREATE OR REPLACE PROCEDURE sp_upsert_dim_pelicula()\n",
    "LANGUAGE plpgsql\n",
    "AS $$\n",
    "BEGIN\n",
    "    -- Insertar o actualizar registros en dim_pelicula\n",
    "    WITH film_base AS (\n",
    "        SELECT\n",
    "            film_id,\n",
    "            title AS titulo,\n",
    "            description AS descripcion,\n",
    "            release_year AS anio_lanzamiento,\n",
    "            language_id,\n",
    "            rental_duration AS duracion_renta,\n",
    "            rental_rate AS tasa_renta,\n",
    "            length AS duracion,\n",
    "            replacement_cost AS costo_reemplazo,\n",
    "            rating AS clasificacion,\n",
    "            special_features AS caracteristicas_especiales,\n",
    "            last_update AS fecha_ultima_actualizacion\n",
    "        FROM pagila.film\n",
    "    ),\n",
    "    idioma AS (\n",
    "        SELECT\n",
    "            language_id,\n",
    "            name AS idioma\n",
    "        FROM pagila.language\n",
    "    ),\n",
    "    categoria AS (\n",
    "        SELECT\n",
    "            fc.film_id,\n",
    "            c.name AS categoria\n",
    "        FROM pagila.film_category AS fc\n",
    "        JOIN pagila.category AS c ON fc.category_id = c.category_id\n",
    "    ),\n",
    "    data_to_upsert AS (\n",
    "        SELECT\n",
    "            fb.film_id,\n",
    "            fb.titulo,\n",
    "            fb.descripcion,\n",
    "            fb.anio_lanzamiento,\n",
    "            i.idioma,\n",
    "            fb.duracion_renta,\n",
    "            fb.tasa_renta,\n",
    "            fb.duracion,\n",
    "            fb.costo_reemplazo,\n",
    "            fb.clasificacion,\n",
    "            fb.caracteristicas_especiales,\n",
    "            c.categoria,\n",
    "            fb.fecha_ultima_actualizacion\n",
    "        FROM film_base AS fb\n",
    "        LEFT JOIN idioma AS i ON fb.language_id = i.language_id\n",
    "        LEFT JOIN categoria AS c ON fb.film_id = c.film_id\n",
    "    )\n",
    "    INSERT INTO dw.dim_pelicula (\n",
    "        film_id,\n",
    "        titulo,\n",
    "        descripcion,\n",
    "        anio_lanzamiento,\n",
    "        idioma,\n",
    "        duracion_renta,\n",
    "        tasa_renta,\n",
    "        duracion,\n",
    "        costo_reemplazo,\n",
    "        clasificacion,\n",
    "        caracteristicas_especiales,\n",
    "        categoria,\n",
    "        fecha_ultima_actualizacion\n",
    "    )\n",
    "    SELECT\n",
    "        film_id,\n",
    "        titulo,\n",
    "        descripcion,\n",
    "        anio_lanzamiento,\n",
    "        idioma,\n",
    "        duracion_renta,\n",
    "        tasa_renta,\n",
    "        duracion,\n",
    "        costo_reemplazo,\n",
    "        clasificacion,\n",
    "        caracteristicas_especiales,\n",
    "        categoria,\n",
    "        fecha_ultima_actualizacion\n",
    "    FROM data_to_upsert\n",
    "    ON CONFLICT (film_id) DO UPDATE\n",
    "    SET\n",
    "        titulo = EXCLUDED.titulo,\n",
    "        descripcion = EXCLUDED.descripcion,\n",
    "        anio_lanzamiento = EXCLUDED.anio_lanzamiento,\n",
    "        idioma = EXCLUDED.idioma,\n",
    "        duracion_renta = EXCLUDED.duracion_renta,\n",
    "        tasa_renta = EXCLUDED.tasa_renta,\n",
    "        duracion = EXCLUDED.duracion,\n",
    "        costo_reemplazo = EXCLUDED.costo_reemplazo,\n",
    "        clasificacion = EXCLUDED.clasificacion,\n",
    "        caracteristicas_especiales = EXCLUDED.caracteristicas_especiales,\n",
    "        categoria = EXCLUDED.categoria,\n",
    "        fecha_ultima_actualizacion = EXCLUDED.fecha_ultima_actualizacion;\n",
    "END;\n",
    "$$;\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_dim_pelicula_upsert)\n",
    "conn_dw.commit()\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_pelicula();\")\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "daabe9f0",
   "metadata": {},
   "source": [
    "### Dimensión Cliente\n",
    "\n",
    "#### Creación de la Tabla con Clave Primaria"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c5bbf6c7",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_create_dim_cliente_table = \"\"\"\n",
    "CREATE TABLE IF NOT EXISTS dw.dim_cliente (\n",
    "    cliente_key SERIAL PRIMARY KEY,\n",
    "    customer_id INTEGER UNIQUE,\n",
    "    nombre TEXT,\n",
    "    apellido TEXT,\n",
    "    email TEXT,\n",
    "    activo BOOLEAN,\n",
    "    direccion TEXT,\n",
    "    ciudad TEXT,\n",
    "    pais TEXT,\n",
    "    fecha_creacion TIMESTAMP,\n",
    "    fecha_ultima_actualizacion TIMESTAMP\n",
    ");\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_create_dim_cliente_table)\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9a269492",
   "metadata": {},
   "source": [
    "#### Procedimiento Almacenado con Upsert"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5f108870",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_dim_cliente_upsert = \"\"\"\n",
    "CREATE OR REPLACE PROCEDURE sp_upsert_dim_cliente()\n",
    "LANGUAGE plpgsql\n",
    "AS $$\n",
    "BEGIN\n",
    "    WITH data_to_upsert AS (\n",
    "        SELECT\n",
    "            c.customer_id,\n",
    "            c.first_name AS nombre,\n",
    "            c.last_name AS apellido,\n",
    "            c.email,\n",
    "            c.active AS activo,\n",
    "            a.address AS direccion,\n",
    "            ci.city AS ciudad,\n",
    "            co.country AS pais,\n",
    "            c.create_date AS fecha_creacion,\n",
    "            c.last_update AS fecha_ultima_actualizacion\n",
    "        FROM pagila.customer AS c\n",
    "        JOIN pagila.address AS a ON c.address_id = a.address_id\n",
    "        JOIN pagila.city AS ci ON a.city_id = ci.city_id\n",
    "        JOIN pagila.country AS co ON ci.country_id = co.country_id\n",
    "    )\n",
    "    INSERT INTO dw.dim_cliente (\n",
    "        customer_id,\n",
    "        nombre,\n",
    "        apellido,\n",
    "        email,\n",
    "        activo,\n",
    "        direccion,\n",
    "        ciudad,\n",
    "        pais,\n",
    "        fecha_creacion,\n",
    "        fecha_ultima_actualizacion\n",
    "    )\n",
    "    SELECT\n",
    "        customer_id,\n",
    "        nombre,\n",
    "        apellido,\n",
    "        email,\n",
    "        activo,\n",
    "        direccion,\n",
    "        ciudad,\n",
    "        pais,\n",
    "        fecha_creacion,\n",
    "        fecha_ultima_actualizacion\n",
    "    FROM data_to_upsert\n",
    "    ON CONFLICT (customer_id) DO UPDATE\n",
    "    SET\n",
    "        nombre = EXCLUDED.nombre,\n",
    "        apellido = EXCLUDED.apellido,\n",
    "        email = EXCLUDED.email,\n",
    "        activo = EXCLUDED.activo,\n",
    "        direccion = EXCLUDED.direccion,\n",
    "        ciudad = EXCLUDED.ciudad,\n",
    "        pais = EXCLUDED.pais,\n",
    "        fecha_creacion = EXCLUDED.fecha_creacion,\n",
    "        fecha_ultima_actualizacion = EXCLUDED.fecha_ultima_actualizacion;\n",
    "END;\n",
    "$$;\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_dim_cliente_upsert)\n",
    "conn_dw.commit()\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_cliente();\")\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7fd849ed",
   "metadata": {},
   "source": [
    "### Dimensión Empleado\n",
    "\n",
    "#### Creación de la Tabla con Clave Primaria"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e4b58ae4",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_create_dim_empleado_table = \"\"\"\n",
    "CREATE TABLE IF NOT EXISTS dw.dim_empleado (\n",
    "    empleado_key SERIAL PRIMARY KEY,\n",
    "    staff_id INTEGER UNIQUE,\n",
    "    nombre TEXT,\n",
    "    apellido TEXT,\n",
    "    email TEXT,\n",
    "    tienda_id INTEGER,\n",
    "    activo BOOLEAN,\n",
    "    nombre_usuario TEXT,\n",
    "    fecha_ultima_actualizacion TIMESTAMP\n",
    ");\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_create_dim_empleado_table)\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d4a8b649",
   "metadata": {},
   "source": [
    "#### Procedimiento Almacenado con Upsert"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9f36dd21",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_dim_empleado_upsert = \"\"\"\n",
    "CREATE OR REPLACE PROCEDURE sp_upsert_dim_empleado()\n",
    "LANGUAGE plpgsql\n",
    "AS $$\n",
    "BEGIN\n",
    "    WITH data_to_upsert AS (\n",
    "        SELECT\n",
    "            s.staff_id,\n",
    "            s.first_name AS nombre,\n",
    "            s.last_name AS apellido,\n",
    "            s.email,\n",
    "            s.store_id AS tienda_id,\n",
    "            s.active AS activo,\n",
    "            s.username AS nombre_usuario,\n",
    "            s.last_update AS fecha_ultima_actualizacion\n",
    "        FROM pagila.staff AS s\n",
    "    )\n",
    "    INSERT INTO dw.dim_empleado (\n",
    "        staff_id,\n",
    "        nombre,\n",
    "        apellido,\n",
    "        email,\n",
    "        tienda_id,\n",
    "        activo,\n",
    "        nombre_usuario,\n",
    "        fecha_ultima_actualizacion\n",
    "    )\n",
    "    SELECT\n",
    "        staff_id,\n",
    "        nombre,\n",
    "        apellido,\n",
    "        email,\n",
    "        tienda_id,\n",
    "        activo,\n",
    "        nombre_usuario,\n",
    "        fecha_ultima_actualizacion\n",
    "    FROM data_to_upsert\n",
    "    ON CONFLICT (staff_id) DO UPDATE\n",
    "    SET\n",
    "        nombre = EXCLUDED.nombre,\n",
    "        apellido = EXCLUDED.apellido,\n",
    "        email = EXCLUDED.email,\n",
    "        tienda_id = EXCLUDED.tienda_id,\n",
    "        activo = EXCLUDED.activo,\n",
    "        nombre_usuario = EXCLUDED.nombre_usuario,\n",
    "        fecha_ultima_actualizacion = EXCLUDED.fecha_ultima_actualizacion;\n",
    "END;\n",
    "$$;\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_dim_empleado_upsert)\n",
    "conn_dw.commit()\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_empleado();\")\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5b0804cf",
   "metadata": {},
   "source": [
    "### Dimensión Tienda\n",
    "\n",
    "#### Creación de la Tabla con Clave Primaria"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eec4632d",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_create_dim_tienda_table = \"\"\"\n",
    "CREATE TABLE IF NOT EXISTS dw.dim_tienda (\n",
    "    tienda_key SERIAL PRIMARY KEY,\n",
    "    store_id INTEGER UNIQUE,\n",
    "    gerente_staff_id INTEGER,\n",
    "    direccion TEXT,\n",
    "    ciudad TEXT,\n",
    "    pais TEXT,\n",
    "    codigo_postal TEXT,\n",
    "    telefono TEXT,\n",
    "    fecha_ultima_actualizacion TIMESTAMP\n",
    ");\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_create_dim_tienda_table)\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9cd2d458",
   "metadata": {},
   "source": [
    "#### Procedimiento Almacenado con Upsert"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ab695e63",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_dim_tienda_upsert = \"\"\"\n",
    "CREATE OR REPLACE PROCEDURE sp_upsert_dim_tienda()\n",
    "LANGUAGE plpgsql\n",
    "AS $$\n",
    "BEGIN\n",
    "    WITH data_to_upsert AS (\n",
    "        SELECT\n",
    "            st.store_id,\n",
    "            st.manager_staff_id AS gerente_staff_id,\n",
    "            a.address AS direccion,\n",
    "            ci.city AS ciudad,\n",
    "            co.country AS pais,\n",
    "            a.postal_code AS codigo_postal,\n",
    "            a.phone AS telefono,\n",
    "            st.last_update AS fecha_ultima_actualizacion\n",
    "        FROM pagila.store AS st\n",
    "        JOIN pagila.address AS a ON st.address_id = a.address_id\n",
    "        JOIN pagila.city AS ci ON a.city_id = ci.city_id\n",
    "        JOIN pagila.country AS co ON ci.country_id = co.country_id\n",
    "    )\n",
    "    INSERT INTO dw.dim_tienda (\n",
    "        store_id,\n",
    "        gerente_staff_id,\n",
    "        direccion,\n",
    "        ciudad,\n",
    "        pais,\n",
    "        codigo_postal,\n",
    "        telefono,\n",
    "        fecha_ultima_actualizacion\n",
    "    )\n",
    "    SELECT\n",
    "        store_id,\n",
    "        gerente_staff_id,\n",
    "        direccion,\n",
    "        ciudad,\n",
    "        pais,\n",
    "        codigo_postal,\n",
    "        telefono,\n",
    "        fecha_ultima_actualizacion\n",
    "    FROM data_to_upsert\n",
    "    ON CONFLICT (store_id) DO UPDATE\n",
    "    SET\n",
    "        gerente_staff_id = EXCLUDED.gerente_staff_id,\n",
    "        direccion = EXCLUDED.direccion,\n",
    "        ciudad = EXCLUDED.ciudad,\n",
    "        pais = EXCLUDED.pais,\n",
    "        codigo_postal = EXCLUDED.codigo_postal,\n",
    "        telefono = EXCLUDED.telefono,\n",
    "        fecha_ultima_actualizacion = EXCLUDED.fecha_ultima_actualizacion;\n",
    "END;\n",
    "$$;\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_dim_tienda_upsert)\n",
    "conn_dw.commit()\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_tienda();\")\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "172af3d5",
   "metadata": {},
   "source": [
    "### Dimensión Tiempo\n",
    "\n",
    "La dimensión tiempo es un poco diferente, ya que se basa en un rango de fechas generado. Podemos asumir que las fechas no cambiarán, pero para mantener consistencia, también aplicaremos upsert.\n",
    "\n",
    "#### Creación de la Tabla con Clave Primaria"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9d3a7af4",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_create_dim_tiempo_table = \"\"\"\n",
    "CREATE TABLE IF NOT EXISTS dw.dim_tiempo (\n",
    "    tiempo_key SERIAL PRIMARY KEY,\n",
    "    fecha DATE UNIQUE,\n",
    "    dia INTEGER,\n",
    "    mes INTEGER,\n",
    "    anio INTEGER,\n",
    "    trimestre INTEGER,\n",
    "    nombre_dia TEXT,\n",
    "    nombre_mes TEXT,\n",
    "    es_fin_de_semana BOOLEAN\n",
    ");\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_create_dim_tiempo_table)\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b03674fd",
   "metadata": {},
   "source": [
    "#### Procedimiento Almacenado con Upsert"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3b6fb75f",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_dim_tiempo_upsert = \"\"\"\n",
    "CREATE OR REPLACE PROCEDURE sp_upsert_dim_tiempo()\n",
    "LANGUAGE plpgsql\n",
    "AS $$\n",
    "BEGIN\n",
    "    WITH date_range AS (\n",
    "        SELECT\n",
    "            MIN(DATE(rental_date)) AS start_date,\n",
    "            MAX(DATE(rental_date)) AS end_date\n",
    "        FROM pagila.rental\n",
    "    ),\n",
    "    dates AS (\n",
    "        SELECT\n",
    "            generate_series(\n",
    "                (SELECT start_date FROM date_range),\n",
    "                (SELECT end_date FROM date_range),\n",
    "                interval '1 day'\n",
    "            )::date AS fecha\n",
    "    ),\n",
    "    data_to_upsert AS (\n",
    "        SELECT\n",
    "            fecha,\n",
    "            EXTRACT(DAY FROM fecha)::INT AS dia,\n",
    "            EXTRACT(MONTH FROM fecha)::INT AS mes,\n",
    "            EXTRACT(YEAR FROM fecha)::INT AS anio,\n",
    "            EXTRACT(QUARTER FROM fecha)::INT AS trimestre,\n",
    "            TO_CHAR(fecha, 'Day') AS nombre_dia,\n",
    "            TO_CHAR(fecha, 'Month') AS nombre_mes,\n",
    "            CASE WHEN EXTRACT(ISODOW FROM fecha) IN (6,7) THEN TRUE ELSE FALSE END AS es_fin_de_semana\n",
    "        FROM dates\n",
    "        ORDER BY fecha\n",
    "    )\n",
    "    INSERT INTO dw.dim_tiempo (\n",
    "        fecha,\n",
    "        dia,\n",
    "        mes,\n",
    "        anio,\n",
    "        trimestre,\n",
    "        nombre_dia,\n",
    "        nombre_mes,\n",
    "        es_fin_de_semana\n",
    "    )\n",
    "    SELECT\n",
    "        fecha,\n",
    "        dia,\n",
    "        mes,\n",
    "        anio,\n",
    "        trimestre,\n",
    "        nombre_dia,\n",
    "        nombre_mes,\n",
    "        es_fin_de_semana\n",
    "    FROM data_to_upsert\n",
    "    ON CONFLICT (fecha) DO UPDATE\n",
    "    SET\n",
    "        dia = EXCLUDED.dia,\n",
    "        mes = EXCLUDED.mes,\n",
    "        anio = EXCLUDED.anio,\n",
    "        trimestre = EXCLUDED.trimestre,\n",
    "        nombre_dia = EXCLUDED.nombre_dia,\n",
    "        nombre_mes = EXCLUDED.nombre_mes,\n",
    "        es_fin_de_semana = EXCLUDED.es_fin_de_semana;\n",
    "END;\n",
    "$$;\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_dim_tiempo_upsert)\n",
    "conn_dw.commit()\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_tiempo();\")\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3f9a51c0",
   "metadata": {},
   "source": [
    "### Tabla de Hechos Alquileres\n",
    "\n",
    "#### Creación de la Tabla con Clave Primaria"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a35a325a",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_create_hechos_alquileres_table = \"\"\"\n",
    "CREATE TABLE IF NOT EXISTS dw.hechos_alquileres (\n",
    "    alquiler_key SERIAL PRIMARY KEY,\n",
    "    rental_id INTEGER UNIQUE,\n",
    "    pelicula_key INTEGER,\n",
    "    cliente_key INTEGER,\n",
    "    empleado_key INTEGER,\n",
    "    tienda_key INTEGER,\n",
    "    tiempo_key INTEGER,\n",
    "    duracion_alquiler NUMERIC,\n",
    "    monto NUMERIC\n",
    ");\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_create_hechos_alquileres_table)\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "42e75200",
   "metadata": {},
   "source": [
    "#### Procedimiento Almacenado con Upsert"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b3b500a",
   "metadata": {},
   "outputs": [],
   "source": [
    "procedure_hechos_alquileres_upsert = \"\"\"\n",
    "CREATE OR REPLACE PROCEDURE sp_upsert_hechos_alquileres()\n",
    "LANGUAGE plpgsql\n",
    "AS $$\n",
    "BEGIN\n",
    "    WITH data_to_upsert AS (\n",
    "        SELECT\n",
    "            r.rental_id,\n",
    "            dp.pelicula_key,\n",
    "            dc.cliente_key,\n",
    "            de.empleado_key,\n",
    "            dt.tienda_key,\n",
    "            dtm.tiempo_key,\n",
    "            EXTRACT(EPOCH FROM (r.return_date - r.rental_date))/60 AS duracion_alquiler,\n",
    "            p.amount AS monto\n",
    "        FROM pagila.rental AS r\n",
    "        JOIN pagila.payment AS p ON r.rental_id = p.rental_id\n",
    "        JOIN pagila.inventory AS i ON r.inventory_id = i.inventory_id\n",
    "        JOIN dw.dim_pelicula AS dp ON i.film_id = dp.film_id\n",
    "        JOIN dw.dim_cliente AS dc ON r.customer_id = dc.customer_id\n",
    "        JOIN dw.dim_empleado AS de ON r.staff_id = de.staff_id\n",
    "        JOIN dw.dim_tienda AS dt ON i.store_id = dt.store_id\n",
    "        JOIN dw.dim_tiempo AS dtm ON DATE(r.rental_date) = dtm.fecha\n",
    "    )\n",
    "    INSERT INTO dw.hechos_alquileres (\n",
    "        rental_id,\n",
    "        pelicula_key,\n",
    "        cliente_key,\n",
    "        empleado_key,\n",
    "        tienda_key,\n",
    "        tiempo_key,\n",
    "        duracion_alquiler,\n",
    "        monto\n",
    "    )\n",
    "    SELECT\n",
    "        rental_id,\n",
    "        pelicula_key,\n",
    "        cliente_key,\n",
    "        empleado_key,\n",
    "        tienda_key,\n",
    "        tiempo_key,\n",
    "        duracion_alquiler,\n",
    "        monto\n",
    "    FROM data_to_upsert\n",
    "    ON CONFLICT (rental_id) DO UPDATE\n",
    "    SET\n",
    "        pelicula_key = EXCLUDED.pelicula_key,\n",
    "        cliente_key = EXCLUDED.cliente_key,\n",
    "        empleado_key = EXCLUDED.empleado_key,\n",
    "        tienda_key = EXCLUDED.tienda_key,\n",
    "        tiempo_key = EXCLUDED.tiempo_key,\n",
    "        duracion_alquiler = EXCLUDED.duracion_alquiler,\n",
    "        monto = EXCLUDED.monto;\n",
    "END;\n",
    "$$;\n",
    "\"\"\"\n",
    "cursor_dw.execute(procedure_hechos_alquileres_upsert)\n",
    "conn_dw.commit()\n",
    "cursor_dw.execute(\"CALL sp_upsert_hechos_alquileres();\")\n",
    "conn_dw.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2ba0ff95",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## Verificación de las Tablas Creadas\n",
    "\n",
    "Verificaremos que las tablas se hayan creado correctamente en la base de datos `pagila_dw` y que contengan los datos esperados."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e54c1917",
   "metadata": {},
   "outputs": [],
   "source": [
    "import psycopg2\n",
    "\n",
    "conn = psycopg2.connect(\n",
    "    dbname=\"pagila_dw\",\n",
    "    user=\"postgres\",\n",
    "    password=\"postgres\",\n",
    "    host=\"localhost\",\n",
    "    port=\"5432\"\n",
    ")\n",
    "\n",
    "cursor = conn.cursor()\n",
    "cursor.execute(\"\"\"\n",
    "    SELECT table_schema, table_name\n",
    "    FROM information_schema.tables\n",
    "    WHERE table_schema = 'dw';\n",
    "\"\"\")\n",
    "tables = cursor.fetchall()\n",
    "for schema, table in tables:\n",
    "    print(f\"Schema: {schema}, Table: {table}\")\n",
    "    \n",
    "cursor.close()\n",
    "conn.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "645d2aa0",
   "metadata": {},
   "source": [
    "### Salida Esperada\n",
    "\n",
    "Esto debería mostrar todas las tablas creadas en el esquema `dw`:\n",
    "\n",
    "```\n",
    "Schema: dw, Table: dim_pelicula\n",
    "Schema: dw, Table: dim_cliente\n",
    "Schema: dw, Table: dim_empleado\n",
    "Schema: dw, Table: dim_tienda\n",
    "Schema: dw, Table: dim_tiempo\n",
    "Schema: dw, Table: hechos_alquileres\n",
    "```\n",
    "\n",
    "## Consultar los Datos\n",
    "\n",
    "Usaremos `ipython-sql` para conectarnos a la base de datos y consultar las tablas creadas."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "56cf0623",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Cargar la extensión SQL\n",
    "%load_ext sql\n",
    "%sql postgresql://postgres:postgres@localhost/pagila_dw\n",
    "\n",
    "# Consultar la tabla dim_pelicula\n",
    "%%sql\n",
    "SELECT * FROM dw.dim_pelicula LIMIT 10;"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d06459d",
   "metadata": {},
   "source": [
    "### Resultado\n",
    "\n",
    "Al ejecutar el código anterior, deberías ver las primeras 10 filas de la tabla `dim_pelicula` en la base de datos `pagila_dw`.\n",
    "\n",
    "---\n",
    "\n",
    "## Explicación de los Cambios Realizados\n",
    "\n",
    "- **Uso de Claves Primarias y Únicas**: Añadimos claves primarias y únicas en las tablas de destino para poder utilizar la cláusula `ON CONFLICT` en las operaciones de inserción.\n",
    "\n",
    "- **Implementación de Upsert**: En lugar de eliminar y volver a crear las tablas, los procedimientos almacenados ahora utilizan `INSERT ... ON CONFLICT DO UPDATE` para insertar nuevos registros y actualizar los existentes basándose en las claves únicas definidas.\n",
    "\n",
    "- **Mantener Datos Históricos**: Al utilizar upsert, podemos mantener los datos existentes y actualizarlos solo si hay cambios, lo que es útil para escenarios donde los datos de origen pueden cambiar con el tiempo.\n",
    "\n",
    "---\n",
    "\n",
    "## Ejecución Regular del ETL\n",
    "\n",
    "Puedes programar la ejecución de estos procedimientos almacenados para que se ejecuten periódicamente y mantengan actualizadas las tablas de tu data warehouse."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc8f0c52",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Llamar a los procedimientos para actualizar los datos\n",
    "cursor_dw = conn_dw.cursor()\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_pelicula();\")\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_cliente();\")\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_empleado();\")\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_tienda();\")\n",
    "cursor_dw.execute(\"CALL sp_upsert_dim_tiempo();\")\n",
    "cursor_dw.execute(\"CALL sp_upsert_hechos_alquileres();\")\n",
    "conn_dw.commit()\n",
    "cursor_dw.close()\n",
    "conn_dw.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6d4c7719",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "Este proceso ha actualizado los procedimientos almacenados para que utilicen **insert-update (upsert)**, permitiendo así que el ETL sea más eficiente y que las tablas en `pagila_dw` reflejen los cambios en los datos de origen sin necesidad de recrearlas cada vez."
   ]
  }
 ],
 "metadata": {
  "jupytext": {
   "cell_metadata_filter": "-all",
   "main_language": "python",
   "notebook_metadata_filter": "-all"
  },
  "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.12.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}