Configuración de base de datos source y entorno para ETL en Python

Después de varios intentos de tratar de conectarme a la base de datos con el Notebook y DBeaver me daban errores, lo que me solucionó fue cambiar de

-p 5432:5432 

a

-p 5433:5432 

(Es lo mismo si tiene un docker-compose)
Imagino que mi instalación local de Postgres sin Docker ya estaba ocupando el puerto 5432, entonces utilicé el 5433.

seria conveniente actualizar este curso algunas cosas no se pueden replicar tal como se indican aca, intente hacerlo paso a paso y no funciona.

No pude configurar la Base de Datos, siempre me pasa que no logro entender este tipo de instrucciones y ahi quedo estancada con la parte practica.

Comparto esta nota, donde destaca lo bueno y lo malo de DataSpell como herramienta.

Para cargar los datos a la tabla les recomiendo dividan el archivo SQL en varios archivos más pequeños e ir cargándolos uno a uno, puesto que el script tiene más de 6 millones de filas y esto puede colgar nuestro pc
Ejemplo de un compañero

Buenas, para los que trabajan con un jupyter notebook y postgres en dbeaver les dejo un script que utilice para :

A partir del archivo completo, generar 200 archivos más pequeños (al original lo trabaje como un .txt en vez de .sql cambiándole la extensión)

Recorro cada archivo .txt y lo envío a postgres (hay que eliminar la última “,” que queda al final de cada archivo como consecuencia de la partición del archivo original

Consideraciones:
*La creación de la tabla la ejecuté directamente desde postgres
*Las sentencias : “CREATE …etc” y "insert into public.trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name)
values " Las eliminé del .txt original para que queden solamente los valores a insertar, luego agrego esa sentencia en cada iteración del loop

import psycopg2  ( previa !pip install psycopg2)
# Establecer los detalles de conexión a la base de datos
host = 'localhost'
database = 'postgres'
port = 5432 (siempre que el puerto queesté usando pos sea 5432, chequear por las dudas, de ser otro, hay que cambiarlo)
username = 'postgres'
password = 'xxxxxxxxx' (reemplazar por la constraseña de cada uno

# Establecer la ruta del archivo completo
sql_file_path = 'postgres_public_trades.txt' (cambié la extensión .sql por .txt y no defino el path del archivo porque está guardado enla misma carpeta que mi .ipynb)

# Leer el contenido del archivo completo
with open(sql_file_path, 'r') asfile:
    sql_statements = file.readlines()

# Calcular la cantidad delíneas que va a tener cada archivo
total_files = 200
lines_per_file = len(sql_statements) // total_files

# Crear los archivos de partes
for i inrange(total_files):
    start_index = i * lines_per_file
    end_index = start_index + lines_per_file

    # Obtener el nombre del archivo de parte
    file_name = f"parte_{i + 1}.txt"

    # Escribir las líneas correspondientes en el archivo de parte
    with open(file_name, "w") as output_file:
        output_file.writelines(sql_statements[start_index:end_index])
## Hasta acá lo que hice fue particionar el archivo original cono 6.000.000 de registros, en 200 archivos más pequeños con unos 30.000 registros en cada uno, aproximadamente. Ahora voy a enviar cada archivo a la base:

# Construir la sentencia SQL de inserción
insert_statement = "insert into public.trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name) values "

# Iterar sobre los archivos desde parte_1.txt hasta parte_200.txt
for i in range(1, 201):
    # Ruta del archivo actual
    file_path = f"parte_{i}.txt"

    # Leer los valores a insertar desde el archivo
    with open(file_path, 'r') asfile:
        values_to_insert = file.read()

    # Eliminar la última coma dela última línea
    values_to_insert = values_to_insert.strip()
    values_to_insert = values_to_insert.rstrip(',')

    # Conectarse a la base de datos
    conn = psycopg2.connect(host=host, database=database, port=port, user=username, password=password)
    cur = conn.cursor()

    # Ejecutar la inserciónde los valores enla base de datos
    cur.execute(insert_statement + values_to_insert)

    # Confirmar los cambios y cerrar la conexión a la base de datos
    conn.commit()
    cur.close()
    conn.close()

Buenas tardes, me gustaría se agregaran las versiones especificas que se están usando para maximizar la reproducibilidad del curso

al correr el contenedor con la imagen postgres va a tomar la que se encuentre como latest, PERO con el tiempo puede llegar a ser incompatible el código, seria bueno que le agreguen el tag de la imagen correcta

Yo lo estoy realizando desde vscode, y lo que hice fue, usar un archivo para dividir el archivo en pequeños archivos y luego usar otro script de python para cargar todos los archivos, les dejo el Repositorio por si quieren revisar y usarlo, puede ser complicado pero se entiende y realmente me funciono, cargue todos los 6 millones de datos en menos de 10 minutos.

Inserta los datos en menos de 3 minutos (dependiendo de tu pc)

He desarrollado un programa para insertar grandes volúmenes de datos en PostgreSQL de manera rápida y eficiente. Surgió como solución a los problemas de rendimiento que enfrentamos al insertar mas de 6 millones de registros.

Me encantaría recibir su feedback para mejorar este proyecto.

Repositorio GitHub

Si no pueden crear “New Workspace Directory”, revisen este video: https://www.youtube.com/watch?v=Ax9jXToHsIo&t=112s

Quien vaya por la opción de postgress + dbveaver dejo una solución para dividir la base de datos con Python, este txt de 6.2Millones de filas que nos proporcionan en el curso.
yo particularmente lo ejecuté desde Jupyter notebook y es sencillo, 3 pasos.
1 crear una copia del txt con el que vamos a trabajar
2 eliminar la primera parte del código SQL , es decir dejar en nuestra copia todo lo que va despues del
insert into public.trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name)
values
ya que esto lo vamos a usar por cada carga que hagamos de cada porción que obtengamos del txt.
3 nuestra copia de 6.2M solo de los valores a insertar, colocar el nombre el el código python y ejecutar (recordar que nuestr txt debe estar en la misma locación/ubicación que nuestro notebook.
Cada uno puede modificar la cantidad de filas por porción, yo las hice de 600mil para obtener 10 partes
splitLen = 621636 # nro que corresponde a la cantidad de filas por cada parte que queremos serparar
outputBase = ‘parte’ # parte.1.txt, parte.2.txt, etc.

leemos nuestro archivo txt

input = open(‘postgres_public_trades_completo.txt’, ‘r’).read().split(’\n’)

at = 1
for lines in range(0, len(input), splitLen):
# creamos una lista con las porciones
outputData = input[lines:lines+splitLen]

# abre el archivo de salida que denominamos anteriormente como "parte" 
# lo escribimos con las lineas seleccionadas y se cierra el archivo con nuevo nombre + nro de archivo
output = open(outputBase + str(at) + '.txt', 'w')
output.write('\n'.join(outputData))
output.close()

# Incrementa el contador
at += 1

Hola, adjunto código para ejecutar los archivos ya divididos, los separé de a 30.000 fueron 208 documentos los cuales se demoraron 8 minutos en insertar los 6 millones de registros.
Recuerden crear un pipenv para las dependencias

import os
import psycopg2
import time

def get_conn():
    conn = psycopg2.connect(database="postgres",
                        host="localhost",
                        user="postgres",
                        password="rootpts",
                        port="5432")
    
    if conn.status == 1:
        cursor = conn.cursor()
        return cursor, conn


def start_script():
    cursor, conn = get_conn()
    print('Starting script...')
    entries = os.listdir('txt/')
    for entry in entries:
        time.sleep(1)
        print(f'Reading {entry}...')
        with open(f'txt/{entry}', 'r') as f:
            data = f.read()
            new_data = data.replace('\n', '')
            without_comma = ''.join(new_data.rsplit(',', 1))
            cursor.execute(f"insert into trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name) VALUES {without_comma}")
            conn.commit()
            print(f'Inserted {entry} into database!')


if __name__ == '__main__':
    start_script()

Finalmente lo logre con DataSpell, me fue complicado crear los ambientes y me toco leer para solucionarlo…

Platzi no es solo ver videos, gracias a los comentarios de esta clase y la comunidad desbloque este logro jeje

Postgres usando un archivo docker-compose

version: "3"
services:
  postgres:
    image: postgres
    container_name: postgres
    ports:
      - "5432:5432"
    volumes:
      - postgres-volume:/var/lib/postgresql/data
    environment:
      - POSTGRES_PASSWORD=mysecretpass
volumes:
  postgres-volume:

Para ejecutar el script use DBeaver

Carga automática de datos por partes

Cuando intenté ingresar los datos directamente (Creación de tabla + 1 INSERT con 6216353 registros directamente) con el archivo “postgres_public_trades.sql” mi computador básicamente casi explota. Con base en lo anterior, decidí realizar un código (notebook, script) en Python que particiona el archivo “postgres_public_trades.sql” para poder realizar las inserciones por partes y en ejecutarlas en serie (partición por partición).

Instrucciones:

1. Crea tu base de datos con el nombre que desees <DB_name>
2. Crea una carpeta con el nombre que desees (ese mismo nombre se lo debes asignar a la variable <folder_name>) y en una ruta de tu preferencia (esa misma ruta se la debes asignar a la variable <parent_dir>), para este ejemplo:

folder_name = "splits"
parent_dir = "C:/Users/HP/Documents/python/002_ETL/"

1. El código está hecho para realizar 20 particiones de la sentencia INSERT y 1 partición para la creación de la tabla. En caso de que quieras cambiar el número de particiones debes cambiar el valor de la variable <n_splits> (Ten en cuenta que el número total de particiones es <n_splits+1>).
   

2. Guarda el archivo “postgres_public_trades.sql” dentro de la carpeta creada en el paso 1 y Descarga el código como notebook o script según tu preferencia (Te recomiendo el notebook si tienes VScode con las extensiones instaladas o cualquier otro IDE que te permita correrlo, el script fue generado a partir del NoteBook) y realiza los cambios necesarios para la ejecución (asignación de folder_name y parent_dir).
   
   

   
   

3. Debes abrir tu IDE (en este caso VScode) o editor de código enlazado con la ruta de la carpeta que creaste y ejecuta el notebook/script (en mi computador, el tiempo de ejecución aproximado fue de 2min para n_splits=20). Luego de ejecutar el código, automáticamente se crea la carpeta splits y el archivo splits.sql dentro de la carpeta creada en el paso 1.
   
   

   
   

   
   

4. Desde la terminal que estés usando para postgres (ya sea desde DataSpell, bash (terminal ubuntu), cmd, PowerShell, psql shell, PSQL Tool de PgAdmin, etc), ejecuta el comando \i seguido de un espacio y la ruta completa para ejecutar el archivo splits.sql (el tiempo de ejecución aproximado fue de 5min). Debes asegurarte de estar conectado en la base de datos (que ya creaste) con el comando \c <DB_name>.

# General Form
\i <parent_dir><folder_name>/splits.sql

# Sample
\i C:/Users/HP/Documents/python/002_ETL/splits.sql

Nota: En mi caso, el tiempo de ejecución (splits con python + carga de datos) total pasó de memoria insuficiente a 7min aproximadamente con el mismo computador 👀. Espero haberte podido ayudar ✌

Error: FATAL: role "postgres" does not exist

• Siguiendo todos los pasos el usuario user=postgres no funcionó.
• Al parecer mi máquina lo creo con el nombre de usuario de la máquina, el cual -para estar seguro- lo puedes obtener en la terminal ejecutando:

> whoami
platzi_user

• Y funcionó el test de conexión: