03_02_ETLs_DAGs.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "03 .02 ETLs-DAGs.ipynb",
      "provenance": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/DiploDatos/AnalisisYCuracion/blob/master/03_02_ETLs_DAGs.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# ETLs y DAGs"
      ],
      "metadata": {
        "id": "EaZExnVOl8HR"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "DISCLAIMER: Esta notebook tiene un proposito mas bien ilustrativo que de implementacion. Es posible que el codigo no logren ejecutarlo exitosamente ya que son fragmentos para mostrar ejemplos.  "
      ],
      "metadata": {
        "id": "81nbzMuxb1gD"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "9Cw1YNN2dXgw"
      },
      "outputs": [],
      "source": [
        "import pandas as pd\n",
        "import datetime as dt\n",
        "import requests\n",
        "import io\n",
        "from sqlalchemy import create_engine, text\n",
        "import os\n",
        "from decouple import config\n",
        "import logging"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# URL del dataset que querramos utilizar\n",
        "URL = 'https://cs.famaf.unc.edu.ar/~mteruel/datasets/diplodatos/sysarmy_survey_2020_processed.csv'"
      ],
      "metadata": {
        "id": "JCF2kQgzdlEL"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Cuando nos queremos conectar a una Base de Datos productiva normalmente necesitamos pasar credenciales para acceder.\n",
        "# Estas credenciales NO deben ser escritas en archivos compartidos subidos a github, sino mas bien en archivos \"privados\".\n",
        "# Una buena practica para manejar credenciales es en archivos \".env\" que solo quedan registrados en su computadora local.\n",
        "# La libreria \"python decouple\" permite leer estos archivos de configuracion .env y manejarlo como variables.  \n",
        "DB_USER = config('DB_USER')\n",
        "DB_PASSWORD = config('DB_PASSWORD')\n",
        "DB_HOST = config('DB_HOST')\n",
        "DB_PORT = config('DB_PORT')\n",
        "\n",
        "# Una buena practica es dejar el codigo de las queries SQL en archivos separados de la notebook, .sql\n",
        "SQL_SCRIPT = 'queries.sql'\n"
      ],
      "metadata": {
        "id": "SGfeF-vodl5Z"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# En lugar de usar prints para ver el avance a medida que va corriendo el script se utilizan los logs.\n",
        "# Los logs basicamente son registros que se van dejando para saber el codigo que ha sido ejecutado.\n",
        "# Es decision arbitraria del programador decidir que desea registrar en los logs.\n",
        "# En python se utiliza la libreria logging https://docs.python.org/3/library/logging.html#logging-levels \n",
        "# La libreria permite definir niveles de logs (ERROR, DEBUG, INFO, etc). Segun la criticidad del error. \n",
        "logger = logging.getLogger(__name__)\n",
        "logger.setLevel(logging.DEBUG)\n",
        "logging.basicConfig(format='%(asctime)s %(levelname)s:%(message)s', level=logging.DEBUG, datefmt='%I:%M:%S')\n"
      ],
      "metadata": {
        "id": "-T-0oW1tdmA5"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "def connection_db():\n",
        "  '''Connect to DB using SQLAlchemy methods. Returns an engine created and connected'''\n",
        "  try:\n",
        "      # ejemplo de conexion a PostgreSQL utilizando SQLalchemy\n",
        "      engine = create_engine(f\"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/survey\".format(),\n",
        "                              echo=False, client_encoding='utf8')\n",
        "      logger.info('Conexion exitosa a la base de datos')\n",
        "      return engine\n",
        "\n",
        "  except ValueError as e:\n",
        "      logger.error(e)"
      ],
      "metadata": {
        "id": "1KI-MwL-dmJ-"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "def extract(url):\n",
        "  # The extract process could be complex including some SQL queries \n",
        "  df = pd.read_csv(url)\n",
        "  logger.info('read_csv exitoso')\n",
        "  return df"
      ],
      "metadata": {
        "id": "rKZG610Eg_ku"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "def transformation(engine):\n",
        "    '''Toma la conexion a la base de datos engine y a partir del sql.script\n",
        "    definido y ejecuta sus queries definidos sobre la base'''\n",
        "    try:\n",
        "        sql_file = open(SQL_SCRIPT)\n",
        "        sql_as_string = sql_file.read()\n",
        "        with engine.connect() as conn:\n",
        "            rs = conn.execute(text(sql_as_string))\n",
        "    except Exception as e:\n",
        "        logger.error(e)"
      ],
      "metadata": {
        "id": "MV0ZOX9NeEuz"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "def load(filename, engine, tablename):\n",
        "    '''Toma el nombre del archivo, la conexion a la base (engine) y el nombre de la tabla.\n",
        "     Crea un dataframe y escribe una tabla a partir del engine\n",
        "    ingestando los datos del archivo en dicha tabla.'''\n",
        "    try:\n",
        "        df['fecha'] = dt.date.today()\n",
        "        df.columns = df.columns.str.lower()\n",
        "        df.to_sql(tablename, con=engine, if_exists=\"replace\")\n",
        "        logger.info('Datos ingestados en la base de datos')\n",
        "        logger.info('Cantidad de registros en el archivo: {}'.format(len(df.index)))\n",
        "    except ValueError as e:\n",
        "        logger.error(e)"
      ],
      "metadata": {
        "id": "CmU5_dj0eEmC"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# la funcion main es muy utilizada en scripts python cuando tenemos archivos .py por ejemplo etl.py\n",
        "# al tener la funcion main pueden correr desde la terminal python etl.py y va a ejecutar lo definido en la funcion main\n",
        "def main():\n",
        "\n",
        "    logger.info('Comienza la extraccion')\n",
        "\n",
        "    engine = connection_db()\n",
        "\n",
        "    df = extract(URL)\n",
        "\n",
        "    load(filename, engine)\n",
        "\n",
        "    transformation(engine)\n",
        "\n",
        "if __name__ == \"__main__\":\n",
        "    logger.info('ETL Process Initialized')\n",
        "    main()"
      ],
      "metadata": {
        "id": "DIsj-vPheExW"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Ejemplo de DAG en Airflow"
      ],
      "metadata": {
        "id": "gQ7-IhnZlt1j"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# ejemplo de DAG en Airflow\n",
        "from airflow import DAG\n",
        "\n",
        "with DAG(\n",
        "    'dag_test',\n",
        "    default_args=default_args,\n",
        "    description='DAG ',\n",
        "    schedule_interval=timedelta(hours=1),\n",
        "    start_date=datetime(2022, 1, 26),\n",
        ") as dag:\n",
        "    extraction = PythonOperator(task_id='extraction',\n",
        "                               python_callable=extract)  # Consulta SQL\n",
        "    transformation = PythonOperator(task_id='transformation',\n",
        "                                    python_callable=transform)    # Procesar datos con pandas\n",
        "    load = PythonOperator(task_id='load',\n",
        "                                 python_callable=load) # Carga de datos \n",
        "\n",
        "\n",
        "    extraction >> transformation >> load"
      ],
      "metadata": {
        "id": "_kli2aoreEz0"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        ""
      ],
      "metadata": {
        "id": "KieKM5AkeE2a"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        ""
      ],
      "metadata": {
        "id": "sEKZV4YEeE_D"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        ""
      ],
      "metadata": {
        "id": "CHa4zJjoeFB1"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}