Data-Pipeline-Dockerizing

Dockerizing en el contexto de Docker se refiere al proceso de crear, configurar y desplegar aplicaciones dentro de contenedores Docker.

Este Lab tiene como objetivo establecer un flujo de trabajo automatizado para la ingestión de datos desde un archivo Parquet, que contiene información sobre viajes en taxi, a una base de datos PostgreSQL. A través del uso de Python y Docker, se desarrollará un script que descargará los datos desde una fuente en línea, los transformará en un formato adecuado y los insertará en la base de datos, optimizando así el manejo de grandes volúmenes de información.

El objetivo final de todo el proceso es permitir la ingestión de datos desde un archivo Parquet a una base de datos PostgreSQL de manera eficiente y automatizada. Aquí tienes un resumen de los objetivos específicos en cada parte del flujo:

1. Descarga de datos:

   • Obtener datos de viajes en taxi desde un archivo Parquet accesible en línea.

2. Transformación de datos:

   • Leer el archivo Parquet y transformarlo en un formato adecuado para su inserción en la base de datos (utilizando pandas).

3. Almacenamiento de datos:

   • Conectar y almacenar esos datos en una base de datos PostgreSQL, facilitando su posterior análisis y uso.

4. Contenerización:

   • Usar Docker para encapsular toda la aplicación en un contenedor, lo que permite facilitar la implementación y garantizar que el entorno de ejecución sea consistente, independientemente de dónde se ejecute.

5. Automatización:

   • Permitir que el proceso de descarga e inserción de datos se realice de forma automática, lo que ahorra tiempo y minimiza errores manuales.

6. Primero crearemos el folder Module-2 en Project-1

cd repos/Boost/Project-1
mkdir Module-2

2. Ahora crearemos en la carpeta Module-2 el archivo Python 4-ingest-data-parquet.py y escribiremos el siguiente código:

note

Los archivos Parquet incluyen metadatos que ya definen los tipos de datos, por lo que no es necesario convertir las columnas tpep_pickup_datetime y tpep_dropoff_datetime al tipo datetime con pandas, como se hizo con el CSV en el Module-1, ingesting-ny-taxi-data-to-postgres.

Code

import os
import sys
from dotenv import load_dotenv
import requests
import pyarrow.parquet as pq
from sqlalchemy import create_engine

# Cargar las variables de entorno
load_dotenv()

def main():
    try:
        user = os.getenv('PG_USER')
        password = os.getenv('PG_PASSWORD')
        host = os.getenv('PG_HOST')
        port = os.getenv('PG_PORT')
        db = os.getenv('PG_DB')
        table_name = os.getenv('PG_TABLE_NAME')
        parquet_name = 'output.parquet'
        parquet_url = os.getenv('PG_URL')

        # Descargar el archivo Parquet
        try:
            response = requests.get(parquet_url)
            response.raise_for_status()  # Lanza un error si la respuesta no es 200
            with open(parquet_name, 'wb') as f:
                f.write(response.content)
        except requests.RequestException as e:
            print(f"Error al descargar el archivo Parquet: {e}")
            sys.exit(1)

        # Crear la conexión a la base de datos
        try:
            engine = create_engine(f'postgresql://{user}:{password}@{host}:{port}/{db}')
        except Exception as e:
            print(f"Error al conectar a la base de datos: {e}")
            sys.exit(1)

        # Leer el archivo Parquet
        parquet_file = pq.ParquetFile(parquet_name)

        # Iterar sobre los batches
        for batch_num, batch in enumerate(parquet_file.iter_batches(batch_size=100000)):
            # Convertir el batch a un DataFrame de pandas
            df = batch.to_pandas()

            # Crear la tabla en la base de datos (solo en el primer batch)
            if batch_num == 0:
                df.head(n=0).to_sql(name=table_name, con=engine, if_exists='replace')

            # Insertar el DataFrame en la base de datos
            df.to_sql(name=table_name, con=engine, if_exists='append', method='multi')

            print(f"Inserted batch {batch_num + 1}")

        print("Finished ingesting data into the PostgreSQL database")

    except Exception as e:
        print(f"Ocurrió un error: {e}")
        sys.exit(1)

if __name__ == '__main__':
    main()

3. Ahora, desde Visual Studio Code, crearemos un segundo archivo llamado .env en la carpeta Module-2

 # Credenciales de la base de datos PostgreSQL
PG_USER=root
PG_PASSWORD=root
PG_HOST=postgres-container
PG_PORT=5432
PG_DB=ny_taxi
PG_TABLE_NAME=yellow_taxi_data

# URL del archivo Parquet
PG_URL=https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-01.parquet

4. Pero antes de crear el contenedor docker, vamos a copiar el contenido de la carpeta Module-2 a la carpeta Module-1/data para probar con el entorno virtual de Python creado en el Module-1. Para lograr esto, vamos a copiar los siguientes comandos

cp ~/repos/Boost/Project-1/Module-2/{*,.*} ~/repos/Boost/Project-1/Module-1/data

Ahora revisemos un poco este comando Linux que lo utilizamos para copiar archivos. Vamos a desglosarlo:

• cp: Este es el comando que se usa para copiar archivos y directorios.

• ~/repos/Boost/Project-1/Module-2/{*,.*}: Esta parte especifica los archivos que se van a copiar.

  • *: Representa todos los archivos y directorios en Module-1, excluyendo los ocultos (esos que comienzan con un punto).
  • .*: Representa todos los archivos ocultos en el mismo directorio. Esto incluye archivos como .env, .gitignore, etc., pero también incluirá . (el directorio actual) y .. (el directorio padre).
  • {*,.*}: La combinación de ambas expresiones permite seleccionar todos los archivos (incluyendo los ocultos) para la copia.

• ~/repos/Boost/Project-1/Module-1/data: Esta es la ruta de destino donde se copiarán los archivos. Se asume que data es un subdirectorio dentro de Module-1.

5. Ahora ubiquémonos en el directorio donde tenemos instalado el entorno virtual venv, que en este caso es Module-1, utilizando el siguiente comando:

cd ~/repos/Boost/Project-1/Module-1

6. Primero, ejecutemos el contenedor de PostgreSQL para realizar la prueba antes de proceder con la dockerización del pipeline

docker run -it \
  --name postgres-container \
  --network my-network \
  -e POSTGRES_USER="root" \
  -e POSTGRES_PASSWORD="root" \
  -e POSTGRES_DB="ny_taxi" \
  -v $(pwd)/nyc-tlc-data:/var/lib/postgresql/data \
  -p 5432:5432 \
  postgres:13

7. Ahora, en otra terminal, nos ubicaremos de nuevo en ~/repos/Boost/Project-1/Module-1/ para activar el entorno virtual y conectarnos a la base de datos Postgres utilizando pgcli

cd ~/repos/Boost/Project-1/Module-1/
source venv/bin/activate
pgcli -h localhost -p 5432 -u root -d ny_taxi

8. Ahora que ya estamos conectados al contenedor Postgres desde pgcli, ejecutemos el siguiente comando para borrar la tabla yellow_taxi_data y garantizar que nuestro pipeline se ejecutó correctamente.

DROP TABLE yellow_taxi_data;

9. Ahora debemos instalar python-dotenv que es una biblioteca de Python que facilita la carga de variables de entorno desde un archivo .env a tu entorno de ejecución. Es especialmente útil en proyectos donde necesitas manejar configuraciones sensibles, como credenciales de bases de datos, claves de API, y otros parámetros que no deseas incluir directamente en tu código. En otra terminal, nos ubicaremos de nuevo en ~/repos/Boost/Project-1/Module-1/ para activar el entorno virtual e instalar la nueva biblioteca:

cd ~/repos/Boost/Project-1/Module-1/
source venv/bin/activate
pip install python-dotenv

11. Si todo está bien, ahora podremos ejecutar el 4-ingest-data-parquet.py. Para esto, solo ejecutamos el siguiente comando desde la misma terminal:

python3 4-ingest-data-parquet.py

Este proceso suele tomar un tiempo, así que solo debes esperar. Sin embargo, desde la terminal donde está corriendo pgcli, podemos consultar cuántos registros se han cargado en la tabla yellow_taxi_data. Solo debes ejecutar la siguiente consulta SQL:

select COUNT(1) from yellow_taxi_data;

Si todo está bien, verás en la terminal cuántos registros contiene la tabla en su estado actual de carga. Por último, solo debes esperar hasta que la operación se complete. Luego, continuaremos con la dockerización del pipeline.

12. Ahora, antes de continuar, borremos de nuevo la tabla yellow_taxi_data para poder ejecutar el pipeline de datos desde cero nuevamente

DROP TABLE yellow_taxi_data;

13. De nuevo, abrimos una nueva terminal y navegamos al Module-1.

cd ~/repos/Boost/Project-1/Module-2

14. Una vez ubicados en el Module-2, procederemos a crear, desde la interfaz de Visual Studio Code, el siguiente archivo llamado Dockerfile. Y agregaremos el siguiente codigo:

# Usa una imagen base de Python
FROM python:3.12-slim

# Establece el directorio de trabajo
WORKDIR /app

# Copia los archivos necesarios al contenedor
COPY requirements.txt ./
COPY .env ./
COPY . .

# Instala las dependencias
RUN pip install --no-cache-dir -r requirements.txt

# Comando para ejecutar el script
CMD ["python", "4-ingest-data-parquet.py"]

15. Por último, crearemos un nuevo requirements.txt al que agregaremos las siguientes bibliotecas:

psycopg
pandas
datetime
requests
pandasql
urllib3
traceback2
pyarrow
fastparquet
pandas
sqlalchemy
psycopg2-binary
python-dotenv

16. Ahora sí, procedamos a construir nuestra imagen de Docker con el siguiente comando:

docker build -t mypipeline

17. Pero antes de poder ejecutar nuestro nuevo contenedor, debemos crear una nueva red de Docker para que el contenedor del pipeline se pueda comunicar con el contenedor de PostgreSQL:

docker network create pg-network

18. Ahora si procedamos a ejecutar nuestro pipeline corriendo nuestra imagen docker con el siguiente comando:

docker run --network my-network mypipeline

Con esto sucederá exactamente lo mismo que al ejecutar el pipeline en el entorno venv, solo que desde un contenedor de Docker no tienes que hacer nada de esto de activar un entorno. Solo debes correr el pipeline con la imagen de Docker creada y el proceso será exactamente el mismo.

Explicación

Ok, revisemos este código en Python que realiza varias tareas, principalmente relacionadas con la descarga del Parquet y la carga de datos en PostgreSQL. ¿Listo para comenzar?

1 - Imports

Empecemos con la importación de bibliotecas. El código importa algunas bibliotecas al principio. ¿Sabes qué hacen esas importaciones?

import os
import sys
from dotenv import load_dotenv
import requests
import pyarrow.parquet as pq
from sqlalchemy import create_engine

Este sección Importa os y sys para manejar el sistema y las variables de entorno. También dotenv para cargar esas variables, requests para hacer solicitudes HTTP, pyarrow para trabajar con archivos Parquet y sqlalchemy para interactuar con bases de datos.

2 - ¿Qué es load_dotenv()?

Luego, se carga un archivo de variables de entorno con load_dotenv(). ¿Por qué crees que se hace eso?

# Cargar las variables de entorno
load_dotenv()

load_dotenv() es una función de la biblioteca python-dotenv, que permite gestionar la configuración de tu aplicación mediante variables de entorno. Las variables de entorno son un mecanismo estándar en sistemas operativos para almacenar configuraciones que pueden ser utilizadas por aplicaciones y procesos.

¿Cómo funciona?

1. Archivo .env:

   • Normalmente, tienes un archivo llamado .env en el directorio raíz de tu proyecto. Este archivo contiene variables en un formato simple, como CLAVE=valor. Por ejemplo, puedes tener entradas como PG_USER=mi_usuario o PG_PASSWORD=mi_contraseña.

2. Cargar Variables:

   • Al llamar a load_dotenv(), la función busca este archivo y carga las variables definidas en él en el entorno del sistema. Esto significa que las variables se hacen accesibles mediante os.getenv(), como si fueran parte de la configuración del sistema.

3. Acceso a Variables:

   • Una vez que has cargado las variables de entorno, puedes acceder a ellas en tu código usando os.getenv('NOMBRE_DE_LA_VARIABLE'). Por ejemplo, para acceder al usuario de la base de datos, usarías os.getenv('PG_USER').

¿Por qué usar load_dotenv()?

1. Seguridad:

   • Almacenar información sensible en un archivo .env evita que esta información se exponga en el código fuente, lo cual es crucial, especialmente si el código se comparte o se almacena en repositorios públicos.

2. Configuración Flexible:

   • Permite cambiar la configuración de la aplicación sin necesidad de modificar el código. Solo necesitas editar el archivo .env.

3. Facilidad de Uso:

   • La gestión de configuraciones se vuelve más ordenada y mantenible. Puedes tener diferentes archivos .env para distintos entornos (desarrollo, pruebas, producción).

4. Compatibilidad:

   • La carga de variables de entorno es un estándar en muchas aplicaciones, lo que facilita la portabilidad y la integración con otros sistemas y servicios.

3 - def main():

Muy bien. Ahora, dentro de la función main(), se intenta obtener varias variables de entorno. ¿Cuáles son y para qué se usan?

def main():
    try:
        user = os.getenv('PG_USER')
        password = os.getenv('PG_PASSWORD')
        host = os.getenv('PG_HOST')
        port = os.getenv('PG_PORT')
        db = os.getenv('PG_DB')
        table_name = os.getenv('PG_TABLE_NAME')
        parquet_name = 'output.parquet'
        parquet_url = os.getenv('PG_URL')

Se obtienen el usuario, contraseña, host, puerto, nombre de la base de datos y el nombre de la tabla, además de la URL del archivo Parquet que se va a descargar.

Definición de la función main()

La función main() es el punto central del script. Es donde se ejecuta la lógica principal del programa.

Manejo de excepciones con try

El uso de try permite capturar errores que pueden ocurrir durante la ejecución del código. Esto es importante para que, si algo falla (como una variable de entorno que falta o un problema de conexión), el programa no se detenga inesperadamente.

Carga de variables de entorno

1. user = os.getenv('PG_USER'): Aquí, se está intentando obtener el valor de la variable de entorno PG_USER. Si la variable no existe, os.getenv() devolverá None. Esto significa que, si no se establece un usuario válido, la conexión a la base de datos fallará más adelante.

2. password = os.getenv('PG_PASSWORD'): Similar al anterior, se obtiene la contraseña de la base de datos. Es crucial que esta información sea correcta para poder autenticarse.

3. host = os.getenv('PG_HOST'): Se obtiene la dirección del host. Puede ser localhost si la base de datos está en la misma máquina o una dirección IP si está en un servidor remoto.

4. port = os.getenv('PG_PORT'): Este es el puerto que la base de datos utiliza para las conexiones. Por defecto, PostgreSQL utiliza el puerto 5432. Si no se establece, también podría causar problemas de conexión.

5. db = os.getenv('PG_DB'): Aquí se obtiene el nombre de la base de datos que se quiere utilizar. Sin este nombre, no se puede realizar ninguna operación en la base de datos.

6. table_name = os.getenv('PG_TABLE_NAME'): Esta variable define el nombre de la tabla en la que se insertarán los datos. Es importante que esta tabla exista o que el script tenga permiso para crearla.

7. parquet_name = 'output.parquet': Se define un nombre fijo para el archivo Parquet que se descargará. Este nombre se utilizará más adelante para leer el archivo y procesar los datos.

8. parquet_url = os.getenv('PG_URL'): Finalmente, se obtiene la URL desde la que se descargará el archivo Parquet. Este archivo contendrá los datos que se insertarán en la base de datos.

4 - Descargar el archivo Parquet

# Descargar el archivo Parquet
        try:
            response = requests.get(parquet_url)
            response.raise_for_status()  # Lanza un error si la respuesta no es 200
            with open(parquet_name, 'wb') as f:
                f.write(response.content)
        except requests.RequestException as e:
            print(f"Error al descargar el archivo Parquet: {e}")
            sys.exit(1)

Ahora, vamos a desglosar este fragmento de código ¿Qué está haciendo exactamente este código? Este código está diseñado para descargar un archivo Parquet desde una URL. Vamos a verlo paso a paso. Primero, se hace una solicitud a la URL donde se encuentra el archivo usando requests.get(parquet_url).

¿Y qué pasa si la solicitud falla? Ah, esa es una parte importante. Después de hacer la solicitud, se llama a response.raise_for_status(). Esto verifica si la respuesta del servidor fue exitosa. Si no lo fue, por ejemplo, si hubo un error 404 porque la URL no existe, se lanza una excepción. Eso significa que el programa puede manejar ese error en lugar de seguir adelante y fallar sin control.

¿Y qué ocurre si la solicitud es exitosa? Si la respuesta es exitosa, el código abre un archivo local en modo de escritura binaria, que se indica con 'wb'. Esto significa que va a escribir datos binarios, que es lo que contiene un archivo Parquet.

¿Por qué se utiliza with para abrir el archivo? Usar with es una buena práctica porque asegura que el archivo se cierre automáticamente una vez que terminas de escribir en él, incluso si ocurre un error mientras estás escribiendo. Esto ayuda a evitar problemas de archivos no cerrados.

Entonces, ¿qué se escribe en el archivo? Se escribe el contenido de la respuesta de la solicitud, que es el archivo Parquet descargado. Eso se hace con f.write(response.content).

¿Y qué pasa si hay algún error durante este proceso? Si ocurre un error en la solicitud, como problemas de conexión, se captura con except requests.RequestException as e. En ese caso, se imprime un mensaje de error que nos dice que la descarga falló, y el programa se detiene con sys.exit(1). El 1 aquí generalmente indica que hubo un error.

Así que, este bloque de código maneja la descarga del archivo, verifica si hay errores y se asegura de que todo se maneje correctamente.

tip

Hacer esto de manera adecuada es muy importante para que nuestros programas sean robustos y no se rompan ante errores inesperados.

5 - Crear la conexión a la base de datos

 # Crear la conexión a la base de datos

        try:
            engine = create_engine(f'postgresql://{user}:{password}@{host}:{port}/{db}')
        except Exception as e:
            print(f"Error al conectar a la base de datos: {e}")
            sys.exit(1)

¿Qué es lo que hace este bloque de código? Se encarga de establecer una conexión a una base de datos PostgreSQL. Es un paso crucial para que la aplicación pueda interactuar con la base de datos.

¿Cómo se establece esa conexión? Se utiliza la función create_engine() de SQLAlchemy, que es una biblioteca muy popular para trabajar con bases de datos en Python. El argumento que le pasamos a create_engine() es una cadena que contiene toda la información necesaria para conectarse a la base de datos: el usuario, la contraseña, el host, el puerto y el nombre de la base de datos.

Pero, ¿qué pasa si algo sale mal al intentar conectarse? Aquí se usa un otra vez el bloque try para manejarlo. Si algo falla, como un error de autenticación o si el servidor no está disponible, se lanzará una excepción. En ese caso, el código entra en la parte del except.

¿Y qué se hace en el except? Si hay una excepción, se captura y se imprime un mensaje de error que incluye la descripción del problema. Luego, el programa se detiene con sys.exit(1). Esto es útil porque evita que el programa continúe ejecutándose sin una conexión válida a la base de datos.

Entonces, este bloque de código es importante para asegurarse de que la aplicación no solo intente hacer operaciones en la base de datos, sino que lo haga de manera segura y controlada.

tip

Si no podemos conectarnos a la base de datos, es mejor detener el programa de inmediato en lugar de dejarlo correr y esperar que funcione.

6 - Leer el archivo Parquet

# Leer el archivo Parquet

        parquet_file = pq.ParquetFile(parquet_name)

Esta línea utiliza la biblioteca pyarrow para abrir y leer un archivo Parquet, que es un formato de archivo optimizado para el almacenamiento y procesamiento de datos tabulares.

Un archivo Parquet es un formato de almacenamiento columnar, lo que significa que almacena datos en columnas en lugar de en filas. Esto permite una compresión más eficiente y un acceso más rápido a los datos, especialmente cuando se trabaja con grandes volúmenes de información.

Entonces, ¿qué hace pq.ParquetFile(parquet_name)? Al llamar a pq.ParquetFile(parquet_name), se está creando un objeto que representa el archivo Parquet que se ha descargado. Este objeto nos permitirá interactuar con el contenido del archivo y leer los datos que contiene.

¿Qué tipo de operaciones se pueden realizar con este objeto?

Con el objeto parquet_file, puedes realizar varias operaciones, como leer los datos en diferentes formatos, obtener información sobre la estructura del archivo, y, en este caso, iterar sobre los "batches" o lotes de datos para procesarlos en partes más manejables.

¿Y por qué es importante leer el archivo Parquet de esta manera?

Leer el archivo como un objeto permite aprovechar las optimizaciones del formato Parquet. Por ejemplo, puedes acceder solo a las columnas que necesitas, lo que reduce la cantidad de datos que se cargan en memoria, mejorando el rendimiento general.

Entonces, esta línea es esencial para poder trabajar con los datos que están almacenados en ese archivo Parquet. Es el primer paso para extraer los datos y comenzar a procesarlos, ya sea para almacenarlos en una base de datos o para realizar análisis.

7 - Iterar sobre los batches

# Iterar sobre los batches

        for batch_num, batch in enumerate(parquet_file.iter_batches(batch_size=100000)):

            # Convertir el batch a un DataFrame de pandas

            df = batch.to_pandas()

Este código itera sobre un archivo Parquet en lotes, donde cada lote puede contener hasta 100,000 filas de datos. La línea for batch_num, batch in enumerate(parquet_file.iter_batches(batch_size=100000)): inicia esa iteración.

¿Por qué es útil procesar los datos en lotes en lugar de cargar todo el archivo a la vez?

Procesar datos en lotes es muy eficiente, especialmente cuando trabajas con grandes volúmenes de información. Cargar todo el archivo en memoria podría causar problemas de rendimiento o incluso fallos si el archivo es demasiado grande. Al trabajar con lotes, mantienes el uso de memoria más bajo y manejas los datos de manera más controlada.

¿Qué sucede con cada lote una vez que se obtiene?

Cada lote se convierte en un DataFrame de pandas usando df = batch.to_pandas(). Esta conversión es importante porque pandas proporciona muchas herramientas y funciones para manipular y analizar datos de manera efectiva.

¿Hay algo en particular que se aprovecha al convertir a un DataFrame?

Sí, al convertir a un DataFrame, puedes realizar operaciones de análisis de datos, filtrado, agrupamiento, y muchas otras funciones que pandas ofrece. Esto simplifica significativamente el trabajo con los datos en comparación con trabajar con ellos en su formato original.

Entonces, cada vez que se procesa un lote, se tiene la flexibilidad de usar todas las funcionalidades de pandas. Esto permite una gran versatilidad al trabajar con los datos, ya que puedes aplicar transformaciones, realizar cálculos, y preparar los datos para insertarlos en la base de datos o para cualquier otro análisis posterior.

¿Y qué sucede después de convertir el lote en un DataFrame?

Después de esta conversión, normalmente se llevarían a cabo más operaciones, como insertar esos datos en una base de datos o realizar análisis adicionales. Todo depende del flujo de trabajo que estés implementando.

Este fragmento de código es clave para la preparación de datos antes de realizar cualquier operación adicional. Es un paso fundamental en el proceso de ingestión y análisis de datos.

8 - Crear la tabla en la base de datos (solo en el primer batch)

# Crear la tabla en la base de datos (solo en el primer batch)

            if batch_num == 0:

                df.head(n=0).to_sql(name=table_name, con=engine, if_exists='replace')

Esta otra línea está comprobando si estamos procesando el primer lote de datos. La variable batch_num se incrementa automáticamente con cada iteración del bucle, así que cuando es igual a 0, significa que es el primer lote.

¿Por qué es importante crear la tabla solo en el primer lote?

Es importante porque normalmente solo necesitas crear la tabla una vez antes de comenzar a insertar datos. Si intentaras crear la tabla en cada lote, podrías generar errores o sobrescribir datos. Por eso, al hacerlo solo en el primer lote, aseguras que la tabla esté lista antes de comenzar la inserción.

¿Qué hace exactamente df.head(n=0).to_sql(...)?

Esa línea utiliza df.head(n=0) para obtener la estructura del DataFrame sin datos. Es una manera de asegurarte de que se crea la tabla con las mismas columnas que tiene el DataFrame. Luego, se usa to_sql(...) para crear la tabla en la base de datos.

¿Qué significa el argumento if_exists='replace'?

Este argumento indica que si la tabla ya existe en la base de datos, debe ser reemplazada. Esto es útil si deseas asegurarte de que la tabla esté actualizada con la estructura correcta antes de insertar nuevos datos. Si no quisieras reemplazarla, podrías usar otras opciones como 'append' para agregar datos a la tabla existente sin sobrescribirla.

note

Es un paso crucial para preparar la base de datos antes de la ingestión de datos. Al hacerlo de esta manera, se minimizan los errores y se asegura que la estructura de la tabla coincida con los datos que vas a insertar.

9 - Insertar el DataFrame en la base de datos

# Insertar el DataFrame en la base de datos

            df.to_sql(name=table_name, con=engine, if_exists='append', method='multi')

Esa línea se encarga de insertar el DataFrame df en la base de datos. Utiliza el método to_sql() de pandas, que es una forma sencilla de transferir datos desde un DataFrame a una tabla en una base de datos SQL.

¿Qué significa el argumento name=table_name?

Este argumento indica el nombre de la tabla en la que deseas insertar los datos. table_name es una variable que contiene ese nombre, y es importante asegurarte de que corresponde a la tabla en la base de datos.

Y, ¿qué hay del argumento con=engine? El argumento con=engine establece la conexión a la base de datos. engine es el objeto que creamos anteriormente, y le dice a to_sql() a qué base de datos se debe conectar para realizar la inserción.

¿Qué hace el argumento if_exists='append'?

Este argumento le dice a la función que, si la tabla ya existe, debe agregar los nuevos datos en lugar de sobrescribir la tabla completa. Esto es útil cuando quieres seguir añadiendo información a una tabla existente sin perder los datos que ya están allí.

¿Y el method='multi'? ¿Cuál es su propósito? Usar method='multi' permite que múltiples filas se inserten en la base de datos a la vez, en lugar de hacerlo fila por fila. Esto puede ser mucho más eficiente y rápido, especialmente cuando estás trabajando con grandes volúmenes de datos.

note

Es una parte crucial del proceso de ingestión de datos, asegurando que los datos se añadan correctamente a la base de datos, y lo hace de una manera que optimiza el rendimiento.

10 - Print

     print(f"Inserted batch {batch_num + 1}")

  

        print("Finished ingesting data into the PostgreSQL database")

¿Qué está haciendo esta línea que dice print(f"Inserted batch {batch_num + 1}")?

Esa línea imprime un mensaje en la consola que indica que se ha insertado un lote de datos en la base de datos. Al usar batch_num + 1, se muestra un número que empieza en 1 en lugar de 0, lo que es más intuitivo para las personas.

Imprimir mensajes de estado es muy útil para el monitoreo y la depuración. Te permite ver el progreso del proceso de ingestión y asegurarte de que los datos se estén insertando correctamente. También puede ayudarte a identificar en qué lote podría ocurrir un problema si algo sale mal.

Y luego está esta otra línea que dice print("Finished ingesting data into the PostgreSQL database"). ¿Qué significa eso? Esa línea se imprime una vez que se han procesado todos los lotes y se ha completado el proceso de ingestión de datos. Es un mensaje de confirmación que indica que todo ha terminado y que el proceso se completó con éxito.

tip

Tener mensajes claros en la consola es una buena práctica, ya que proporciona una retroalimentación útil al usuario o al desarrollador, lo que facilita el seguimiento del flujo de trabajo y la identificación de problemas si surgen.

11 - Except

except Exception as e:

        print(f"Ocurrió un error: {e}")

        sys.exit(1)

Este bloque se encarga de manejar excepciones, es decir, cualquier error que pueda ocurrir en el código que se encuentra dentro del bloque try anterior. Si se produce un error, el flujo de ejecución se transfiere a esta sección.

¿Y qué sucede cuando se captura un error?

Cuando se captura un error, se almacena en la variable e. Luego, la línea print(f"Ocurrió un error: {e}") imprime un mensaje en la consola que incluye información sobre el error que ocurrió. Esto es útil para entender qué salió mal.

¿Por qué es importante manejar errores de esta manera?

Manejar errores es crucial para crear aplicaciones robustas. Si no se gestionan, los errores pueden causar que el programa se detenga abruptamente, lo que puede llevar a la pérdida de datos o a un estado inconsistente. Al capturar y manejar excepciones, puedes proporcionar retroalimentación clara al usuario y tomar decisiones sobre cómo proceder.

Y luego, ¿qué significa sys.exit(1)?

Llamar a sys.exit(1) finaliza el programa. El 1 generalmente indica que ha habido un error, mientras que 0 se usaría para indicar una finalización exitosa. Esto es útil porque garantiza que el programa no continúe ejecutándose si hay un problema crítico que debe ser resuelto primero.

tip

Es una buena práctica incluir manejo de excepciones en tu código para mejorar la robustez y la claridad.

12 - if name == 'main'

if __name__ == '__main__':

    main()

¿Qué significa esta línea if __name__ == '__main__':?

Esa línea se usa para comprobar si el script está siendo ejecutado directamente o si está siendo importado como un módulo en otro script. Cuando un archivo de Python se ejecuta, la variable especial __name__ se establece en '__main__'. Si esta condición es verdadera, significa que el script se está ejecutando directamente.

¿Por qué es útil hacer esta comprobación?

Es útil porque permite que el código dentro del bloque se ejecute solo cuando el script se está ejecutando de forma independiente. Si alguien importara este script en otro archivo, el bloque main() no se ejecutaría automáticamente. Esto ayuda a evitar que el código se ejecute sin querer cuando simplemente estás utilizando funciones o clases definidas en este archivo.

Y la línea main() que sigue, ¿qué hace?

Llama a la función main(), que es donde está implementado el flujo principal de la aplicación. Todo el procesamiento y la lógica del programa se encuentran dentro de esa función.

Entonces, este bloque asegura que el código se ejecute de manera adecuada dependiendo de cómo se esté utilizando el script. Es una buena práctica en Python incluir esta estructura, ya que facilita la reutilización del código y mejora la organización.