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

Clase 6 de 25 • Curso de Fundamentos de ETL con Python y Pentaho

¡Hola! En esta clase conocerás cómo configurar una base de datos con SQL, que será una de las 3 fuentes para extraer datos en el proyecto de ETL. Las otras dos fuentes son un archivo JSON y otro CSV que conocerás en clases posteriores.

Además, conocerás cómo conectarte a esta base de datos OLTP con un software de administración de bases de datos. Puede ser DataSpell, DBeaver o el de tu preferencia.

Te sugiero usar DataSpell. Más adelante de este tutorial verás cómo configurarlo.

💡Algo que tenemos que destacar es que la base de datos SQL source no se tendría que crear en un proceso de ETL. Esta base de datos ya estaría creada en algún lado de la infraestructura de los sistemas y aplicaciones de la empresa donde estés colaborando.

En este caso lo estamos haciendo por fines educativos para que tengas una base de datos de donde tomar datos y conozcas el proceso de extracción.

Para la configuración de nuestra base de datos source usaremos PostgreSQL. Podemos utilizarlo de dos formas, una instalación local de PostgreSQL o una configuración por Docker. Te sugiero hacerlo por Docker.

1. Crear container en Docker

Recordemos que Docker es un entorno de gestión de contenedores, de manera que usaremos una imagen base con toda la configuración que requerimos sin instalar necesariamente en nuestra máquina. Solo utilizando los recursos del sistema para correr dicha imagen, algo similar a una máquina virtual.

Por ahora, solo necesitas haber tomado el Curso de Python: PIP y Entornos Virtuales para conocer lo esencial de cómo usar esta herramienta con Python. En ese curso encontrarás la clase para saber cómo instalarlo en tu computador.

Una vez que tengas instalado Docker en tu computador, ejecuta este comando en tu terminal:

WSL 2, Linux o macOS

sudo docker run -d --name=postgres -p 5432:5432 -v postgres-volume:/var/lib/postgresql/data -e POSTGRES_PASSWORD=mysecretpass postgres

Windows

docker run -d --name=postgres -p 5432:5432 -v postgres-volume:/var/lib/postgresql/data -e POSTGRES_PASSWORD=mysecretpass postgres

Como podrás notar, en este comando se específico lo siguiente para la creación de la base de datos con Docker:

Nombre del container: --name=postgres
Puerto a compartir con la máquina local: -p 5432:5432
Volumen para el manejo de disco e información: -v postgres-volume:/var/lib/postgresql/data
Password en PostgreSQL: POSTGRES_PASSWORD=mysecretpass

1.5 Instalación local de PostgreSQL (opcional)

De no usar Docker podrías ver la clase del curso de PostgreSQL en donde aprendes a instalarlo localmente, pero te sugiero intentarlo con Docker ya que puede agilizar tu flujo de trabajo. 😉

2. Validar container creado

Una vez que hayas creado el container de Docker usa el comando docker ps en tu terminal. Podrás ver todos los contenedores que se encuentran en ejecución actualmente y una descripción.

Deberás ver la IMAGE postgres.

3. Configurar DataSpell

Para conectarte a la base de datos usarás un software de administración de bases de datos. Existen varios que puedes utilizar. Para el seguimiento del curso te sugiero utilizar DataSpell o, en su defecto, DBeaver.

DataSpell es un IDE completo para ciencia de de datos donde, además de conectarte y hacer consultas a bases de datos, podrás ejecutar Jupyter Notebooks. ¡Todo en el mismo lugar! 💪🏽

💡 Una de sus desventajas es que es de pago, pero tiene un período de prueba de 30 días para que lo pruebes con este curso. Además existen ciertas opciones para obtener licencias para estudiantes de bachillerato y universidad.

⚠️🦫 En caso de que decidas usar DBeaver en lugar de DataSpell, utiliza tu entorno local de Jupyter Notebooks con Anaconda para la ejecución del código Python de las siguientes clases. 🐍

Instalación de DataSpell

Para instalar DataSpell ve a su sitio web aquí y descarga la versión para tu sistema operativo.📥
Instálalo siguiendo las instrucciones que te aparezcan en el instalador.

⚠️ Cuando te solicite actualizar PATH Variable acepta marcando la opción que te indique. Esto es para evitar errores de ambientes en el futuro. En Windows se ve así:

Al finalizar te pedirá reiniciar el computador:

Abre DataSpell ya que se haya instalado. Al hacer esto por primera vez te pedirá iniciar sesión. Elige la versión free trial registrando tu cuenta para ello.
Una vez que tengas tu cuenta configurada te pedirá elegir un intérprete de Python 🐍.

Previamente deberás tener instalado Anaconda en tu sistema operativo. Te recomiendo que crees un ambiente de Anaconda (Conda environment) único para el proyecto del curso. Llama al ambiente fundamentos-etl.

Elige el ambiente de Anaconda que usarás para el proyecto y presiona el botón Launch DataSpell.

Elegir un intérprete de Anaconda servirá para ejecutar Jupyter Notebooks en DataSpell.

Crea un nuevo Workspace en DataSpell. Presiona el botón File en la barra superior y luego elige la opción New Workspace Directory.

Llama fundamentos-etl al workspace y presiona el botón azul Create.

Elegir ambiente de WSL2 (opcional si usas WSL)

Si quieres usar DataSpell con tu entorno en Windows con WSL 2, deberás conectar DataSpell al ambiente de Anaconda que tenga tu WSL.🐍

Crea un ambiente de Anaconda en tu WSL dedicado al proyecto de tu curso si todavía no lo has hecho. Llámalo fundamentos-etl

conda create --name fundamentos-etl python=3.9

Después ve a DataSpell en su parte inferior donde aparece el intérprete. Presiona la dirección que aparece y elige la opción Interpreter Settings.

Escoge el workspace fundamentos-etl creado anteriormente en DataSpell.

⚠️OJO: el workspace y el Anaconda Environment no son lo mismo. El Anaconda Environment lo vamos a cargar dentro del Workspace de DataSpell.

Después presiona el botón Add Interpreter e inmediatamente selecciona la opción On WSL.

Elige la distribución de Linux a usar y da clic en el botón Next cuando aparezca el mensaje "Instrospection completed succesfully!

Elige el intérprete a usar. Este puede ser un Virtualvenv Environment, el System Interpreter o un Conda Environment. Elige la opción de Conda Environment.

Mara la casilla Use existing environment. Elige el Conda Environment de WSL que usarás para tu proyecto. Anteriormente debiste crearlo desde tu terminal en WSL y llamarlo fundamentos-etl.

Finalmente, presiona el botón azul Create.

Para terminar el proceso presiona el botón azul OK en la parte inferior.

Listo, ya deberá aparecer tu entorno de Anaconda en WSL cargado en la parte inferior de DataSpell.

⚠️Si te aparece un error que indique que el ambiente no puede ser usado como el intérprete del workspace es porque estás intentando cargar el ambiente en el workspace general y no en un workspace de DataSpell que creaste.

Aquí encuentras la guía oficial de cómo conectar tu DataSpell al intérprete de Python o Anaconda en WSL, por si necesitas aprender a configurarlo a detalle.

Recuerda que otra alternativa en Windows es instalar Anaconda para Windows y conectar DataSpell directamente a esta versión.

4. Conexión a la base de datos PostgreSQL

Sigue estos pasos para conectarte a la base de datos postgres desde DataSpell.

Abre DataSpell en tu computador.

Ve a la pestaña de Database y en ella da clic en el botón de signo de +.

Selecciona la opción de Data Source y dentro del menú desplegable elige la opción de PostgreSQL.

Introduce los datos siguientes en la conexión:

Name: local_postgres
Host: localhost
Port: 5432
User: postgres
Database: postgres
Url (opcional): jdbc:postgresql://localhost:5432/postgres
Password: mysecretpass

Da clic en el botón de Test Connection para probar la conexión. Puede que te solicite actualizar unos drivers, acéptalos. Una vez que indique que la conexión es exitosa, da clic en el botón OK.

Listo, ya tienes tu base de datos conectada en DataSpell.

4. Cargar datos en la base de datos Postgres

Dentro de DataSpell, ya con la conexión a la base de datos previamente creada, ejecutarás el script postgres_public_trades.sql.

Descárgalo aquí de Google Drive. 📥

⚠️Este archivo pesa cerca de 500 MB, por lo que puede demorar su descarga. Contiene la creación de una tabla llamada trades y los insert de registros de la tabla.

⚠️Es posible que al intentar correr este script en DBeaver no sea posible por falta de memoria. Te sugerimos cortarlo en varias partes y cargar cada script independientemente.

Una vez descargado el archivo postgres_public_trades.sql sigue estos pasos para cargar los datos con DataSpell:

Da clic derecho sobre la base de datos de PostgreSQL.

Posteriormente da clic en SQL Script y luego en Run SQL Scripts.

Ubica el script descargado dentro de tu computador y da clic en OK.

⚠️La creación de la tabla y la carga de datos puede demorar cerca de 15-20 minutos en DataSpell.

5. Prueba la tabla trades

Una vez terminada la ejecución del script, consulta la tabla Trades ya cargada. Abre el editor de queries desde tu base de datos en DataSpell e ingresa la siguiente consulta:

SELECT * FROM trades;

¡Listo! Ya tienes lo esencial para comenzar a extraer datos de una base de datos OLTP y correr tus notebooks de Python.

Avanza a la siguiente clase. ⚙️

Alberto Fleitas

student•

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

-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.

Brayan omar cabrera diez

student•

Gracias, tu respuesta me ayudo, este curso está muy desactualizado.

Enrique Alejandro Terrazas Huamaní

student•

Confirmo, me funcionó este cambio

Luis Damián Campana

student•

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

Edwin Uldarico Hernandez Osorio

student•

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

Patricia Calisto Gutierrez

student•

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.

Juan Ramón Salazar Garate

student•

Me pasa lo mismo. Me obliga a utilizar dataspell, no podrían hacer el tutorial añadiendo con JupyterNotebook desde Anaconda, o poner el tutorial de como realizarlo.

Yeder Laura vicente

student•

Puedes usar docker-compose.

Crear un carpeta de trabajo
Crear un archivo con nombre docker-compose.yml
Agregar esto:

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:

Ejecutas en tu consola usando: docker-compose up

En mi caso estoy usando DBeaver

Selecionas Postgres, colocas los datos de la base de datos y te conectas. tw: @yederlvicente

Katherine Canaza

student•

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

Juan Sebastián Vargas Castañeda

student•

👌🏻

Alberto Fleitas

student•

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

Luis Mata

student•

Hola, nohe podido terminar de configurar el ambiente, con WSL2. AL ejecutar el comando conda create --name fundamentos-etl python=3.9 En el ambiente WSL2 Sale el siguiente error conda: command not found

Agradezco cualquier apoyo

Luis Falconí

student•

En los comentarios de esta clase esta el repositorio de Daniel da Silva, créditos a el. Con su explicación logre terminar la configuración, tal vez te sea de utilidad.

https://github.com/ddasilva64/ETLPPT23001esp/blob/master/extraccion-de-datos.md

Eduard Giraldo Martínez

student•

Después de leer mucho y 4 horas de intentos, la solución fue usar el puerto 8000

sudo docker run -d --name=postgres -p 8000:5432 -v postgres-volume:/var/lib/postgresql/data -e POSTGRES_PASSWORD=mysecretpass postgres

Liliana Andrea Bobadilla Duran

student•

Gracias, llevaba ya más de 3 horas con errores de conexión. Esta solución me salvó la tarde. Gracias. :)

Maximiliano Vazquez Massaro

student•

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()

Percy Oswaldo Herrera Mogrovejo

student•

Gracias por tu aporte Maximiliano, me ayudo muchísimo para poder importar los registros a Postgres.

Percy Oswaldo Herrera Mogrovejo

student•

Olvide comentar que con este procedimiento ya no será necesario instalar ni configurar DatSpell.

ROGER DIAZ

student•

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.

andres maldonado

student•

Excelente aporte, gracias a ti pude ejecutar todo en vscode ahorrandome instalar mil cosas que para mi no son necesarias en mi día a día

Carlos Eduardo Bracho Rosales

student•

tengo un error OutOfMemory, no he podido cargar la base de datos a postgresql ☹️. lo intente usando postgres en un contenedor de docker y usando dbeaver instalado localmente.

Tambien lo intente dentro del mismo contenedor de postgres cargando el archivo desde la terminal, pero mi computadora se queda paralizada.

Como lo han hecho ustedes? que sugerencia me dan?

Camilo Granda Gómez

student•

Ve a C:\Users\USER\AppData\Local\DBeaver
Abre dbeaver.ini
Cambia -Xmx1024m a -Xmx4096m

A mi me deja de salir ese error, sin embargo se queda colgado mi pc cargando el archivo, jaja.

Carlos Eduardo Bracho Rosales

student•

Gracias por tu respuesta y Camilo pero justo en este momento lo acabo de resolver con un método mas "rudimentario". Dividí el archivo SQL en varias partes iguales (20 para ser exacto) usando el comando split:

split -n 20 -verbose postgres_public_trades.sql

Después tuve que modificar todos esos archivos para que no hubiera errores en las datos usando el editor de codigo. Cuando termine, pase todos esos archivos al contenedor de postgres y los importe. La computadora ya no se me congelo porque eran archivos mas pequeños 😁. Espero no tener mas problemas en lo que queda del curso 😅

Manuel Alejandro Orejuela Garcés

student•

SOLUCIÓN ERROR DE CONEXIÓN BASE DE DATO

Compañeros, en pocas palabras el error se produce porque el puerto 5432 esta ocupado por otros serivicios de nuestro computador, ya sea otras aplicaciones en segundo plano. Se puede producir por instalaciones pasadas que hayan realizado de postgreSQL que genere que diferentes instancias de posgre.exe esten en ejecución al mismo tiempo. Deben de liberar el puerto 5432, que otros servicios no lo esten usando o en su defento cambien la configuración del contenedo ren docker al crearlo, configurando un puerto distinto:

ejemplo:

docker run -d --name=postgres -p 5433:5433 -v postgres-volume:/var/lib/postgresql/data -e POSTGRES_PASSWORD=mysecretpass postgres

si desean verificar si su puerto 5432 esta en uso, este comando les será de utilidad:

tasklist | find "postgres" : les ayudará a verificar si existen muchas instancias postgres.exe en ejecución a la vez.

netstat -an | find "5432" : Les permitirá ver que tan congestionado esta su puerto 5432 y si es necesario usar otro puerto o detener servicios que ya no deseen usar.

Espero que les sea de utilidad.

martin chavez

student•

duda, puedo usar vscode asi directo directo en lugar de ds?

Joshua Paternina

student•

Buenas comunidad Platzi hay alguna otra forma que no sea utilizando DataSpell, ya que no eh me deja ni crear la BD y llevo días intentándolo y quisiera seguir con el curso, muchas gracias

Alarcon7a

student•

puedes usar cualquier motor como jupyter notebook, https://platzi.com/cursos/entorno-ciencia-datos/

Andrés González

student•

Como yo ya tenía mi postgres instalado tuve varios problemas, pero como quería seguir el curso al pie de la letra me toco hace un cambio en el puerto para poder crear mi imagen de postgress

sudo docker run -d --name=postgres -p 5433:5432 -v postgres-volume:/var/lib/postgresql/data -e POSTGRES_PASSWORD=mysecretpass postgres

Esteban Daniel Mares Flores

student•

Con este queda en menos de 5 minutos

import re, psycopg2
from pathlib import Path

# Variables

# Conección postgres
db_config = {
    "host": "localhost",
    "port": "5432",
    "database": "postgres",
    "user": "postgres",
    "password": "mysecretpass"
}

conn = psycopg2.connect(**db_config)
cur = conn.cursor()


# Directorio para los fragmentos
p = Path('frag_sql')

p.mkdir(exist_ok=True)

print("Iniciando fragmentación y carga")

with open('postgres_public_trades.sql', 'r') as op:
    # Obtener número de filas
    for count, line in enumerate(op):
        pass
    
    filas_tot = count + 1
    partition = 20000
    repeticiones =  (filas_tot) // partition
    final = (filas_tot) % partition


with open('postgres_public_trades.sql', 'r') as op:

    for i in range(repeticiones): # archivos a crear
        
        if i == 0: # Primer bloque (ya tiene 'insert into')
            chunck = re.sub(',$', ';', ''.join([next(op) for _ in range(partition)]))

        elif i == repeticiones - 1: # Último bloque (menos repeticiones)
            chunck = re.sub(',$', ';', ''.join([next(op) for _ in range(final)]))
            chunck = f'''insert into public.trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name)
values
{chunck}'''
        else:
            chunck = re.sub(',$', ';', ''.join([next(op) for _ in range(partition)]))
            chunck = f'''insert into public.trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name)
values
{chunck}'''

        # Crear archivos sql
        with open(Path.joinpath(p, f'frag_{i+1}.sql'), 'w') as sql:
            sql.write(chunck)

        # Correr la consulta SQL
        try:
            cur.execute(chunck)
        except psycopg2.Error as e:
            print(f'frag_{i+1}.sql No se cargó correctamente')

# Confirmar la transacción y cerrar la conexión
conn.commit()
conn.close()

print("Fragmentos creados y cargados")

Christian Yael Laurean Martinez

student•

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

Elkin Rodriguez

student•

Luego de muchos intentos logre la conexión con DBeaver, cambiando el puerto al 5433, ya que el 5432 ya estaba en uso en mi equipo local. Básicamente, por si les es de utilidad, deben crear la imagen en Docker así: ```js docker run -d --name=postgres -p 5433:5432 -v postgres-volume:/var/lib/postgresql/data -e POSTGRES_PASSWORD=mysecretpass postgres

nicolas machare

student•

para utilizar este metodo es importante dejar como password "mysecretpass" si no se puede armar una confusion con las contraseñas. Si se quiere utilizar DataSpell y estas teniendo problemas con la conexion a la base de datos, DBeaver es mejor para probar la conexion porque tiene mas soporte en internet. Me sumo al comentario del compañero para no agregar un nuevo post sobre el mismo tema. Saludos

Davies

student•

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

Luis Damián Campana

student•

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

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:

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()

import re, psycopg2
from pathlib import Path

# Variables

# Conección postgres
db_config = {
    "host": "localhost",
    "port": "5432",
    "database": "postgres",
    "user": "postgres",
    "password": "mysecretpass"
}

conn = psycopg2.connect(**db_config)
cur = conn.cursor()


# Directorio para los fragmentos
p = Path('frag_sql')

p.mkdir(exist_ok=True)

print("Iniciando fragmentación y carga")

with open('postgres_public_trades.sql', 'r') as op:
    # Obtener número de filas
    for count, line in enumerate(op):
        pass
    
    filas_tot = count + 1
    partition = 20000
    repeticiones =  (filas_tot) // partition
    final = (filas_tot) % partition


with open('postgres_public_trades.sql', 'r') as op:

    for i in range(repeticiones): # archivos a crear
        
        if i == 0: # Primer bloque (ya tiene 'insert into')
            chunck = re.sub(',$', ';', ''.join([next(op) for _ in range(partition)]))

        elif i == repeticiones - 1: # Último bloque (menos repeticiones)
            chunck = re.sub(',$', ';', ''.join([next(op) for _ in range(final)]))
            chunck = f'''insert into public.trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name)
values
{chunck}'''
        else:
            chunck = re.sub(',$', ';', ''.join([next(op) for _ in range(partition)]))
            chunck = f'''insert into public.trades (country_code, year, comm_code, flow, trade_usd, kg, quantity, quantity_name)
values
{chunck}'''

        # Crear archivos sql
        with open(Path.joinpath(p, f'frag_{i+1}.sql'), 'w') as sql:
            sql.write(chunck)

        # Correr la consulta SQL
        try:
            cur.execute(chunck)
        except psycopg2.Error as e:
            print(f'frag_{i+1}.sql No se cargó correctamente')

# Confirmar la transacción y cerrar la conexión
conn.commit()
conn.close()

print("Fragmentos creados y cargados")

# 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

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

Introducción a ETL

¿Qué es un ETL en ingeniería de datos?

Conceptos base de ETL

Consideraciones de ETL

Servicios y herramientas para ETL

Extracción de datos

Sources

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

Extracción de datos con Python y Pandas

Transformación de datos

Transformación

Transformación de datos con Python y Pandas

Transformación de datos de países

Transformación de datos de transacciones

Carga de datos

Carga

Configuración de clúster en AWS Redshift

Crear tablas en AWS Redshift

Carga de datos con Python

Estructurar sentence para carga de datos de S3 a Redshift

Carga de datos: subida de archivos a AWS Redshift

ETL con Pentaho

Instalación de Pentaho

Extracción de datos con Pentaho

Transformación de datos con Pentaho

Transformación de datos con Pentaho: parte 2

Transformación de datos con Pentaho: parte 3

Carga de datos con Pentaho

Conclusión

Siguientes pasos

Comparte tu proyecto de ETL y obtén tu certificado