Transformaciones de Datos ETL con Python y MySQL

Clase 20 de 24 • Curso de ETL e ingesta de datos con Python

Contenido del curso

Fundamentos de ETL con Python

Técnicas Efectivas de Transformación de Datos

Carga de Datos y Proyecto Final

Resumen

¿Qué vamos a hacer en este proyecto de ETL?

Prepara tus herramientas y conocimientos de Python y MySQL, porque en este proyecto llevaremos a cabo un proceso completo de Extracción, Transformación y Carga (ETL). Utilizaremos como base nuestra base de datos existente llamada "Shakira". Las transformaciones a realizar son únicas y permitirán optimizar y reorganizar los datos de la tabla de actores. Al finalizar, todo lo que transformemos se transferirá a una nueva base de datos, "Shakira ETL". Descubramos los emocionantes pasos que realizaremos a continuación.

¿Cuáles son las transformaciones clave del proyecto?

Filtrado de actores: Identificaremos aquellos actores cuyo nombre empieza con la letra 'a'.
Creación de una columna completa para el nombre: Concatenaremos las columnas first name y last name para formar la columna full name.
Calcular la longitud del nombre: Crearemos namelends, que reflejará la longitud total del nombre.
Conversiones a mayúsculas: Todos los nombres se convertirán a mayúsculas.
Filtrado por longitud del nombre: Mantendremos actores cuyo full name sea mayor de diez caracteres.
Cálculo del año de registro: Determinaremos el año en el cual cada actor se registró.
Agrupación por apellido: Agruparemos los datos por last name y contaremos cuántos actores comparten el mismo apellido.
Indicador de nombre único: Añadiremos una columna que marca si un actor tiene un nombre único.
Estatus del actor: Se agregará un estatus para indicar si el actor tiene más de diez años en la base de datos.
Eliminación de una columna innecesaria: Finalmente, se eliminará la columna last update.

¿Cómo nos conectamos a MySQL con Python?

Para comenzar, debemos garantizar una conexión estable con MySQL. Aquí están los pasos para conectarnos exitosamente:

Importación de librerías necesarias:

MySQL Connector: Para manejar la conexión con MySQL.
Pandas: Para realizar transformaciones de datos.
SQLAlchemy: Para ayudar en la carga de datos.

datetime: Para trabajar con campos de fecha.

import mysql.connector
import pandas as pd
import sqlalchemy
from datetime import datetime

Configuración de la conexión:

Se utiliza "localhost" para servidores locales.
Proporciona tus credenciales como usuario y contraseña.

Conéctate a la base de datos "Shakira".

connection = mysql.connector.connect(
    host='localhost',
    user='root',
    password='yourpassword',
    database='Shakira'
)
cursor = connection.cursor()

Extracción de datos:

Utilizamos una consulta SQL simple para obtener todos los campos de la tabla de actores.

Transformamos los resultados en un data frame de Pandas para su manipulación.

query = "SELECT * FROM actor"
cursor.execute(query)
results = cursor.fetchall()
columns = [col[0] for col in cursor.description]
df_actors = pd.DataFrame(results, columns=columns)

Seguridad de conexiones:

Siempre cierra la conexión cuando no es necesaria para evitar problemas de seguridad.
```
connection.close()
```

Ahora que hemos establecido la conexión y extraído los datos iniciales de MySQL usando Python, tenemos la base perfecta para iniciar las transformaciones. Si encontraste algún obstáculo, no dudes en compartirlo y juntos encontraremos la solución. Adelante, queda mucho por explorar y aprender en próximas etapas. ¡Nos vemos en el siguiente reto ETL!

Iris Aguilar

student•

Debí hacer varios ajustes para lograr la conexión a la Basa de Datos. No me funcionó mysql.conector, aún cuando actualicé las librerías por lo que terminé usando pymsql y así quedó mi código:

No sé si sea correcto pero me funciona perfectamente.

import mysql.connector, pymysql
import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime
from sqlalchemy import text
pd.set_option('mode.chained_assignment', None)

# String de conexión: MySQL 
engine = create_engine("mysql+pymysql://root:12345678@localhost/sakila")
# Probar conexión 
connection = engine.connect() 
print("Conexión exitosa a MySQL", type(connection))

query = 'select * from actor'
data = connection.execute(text(query))

df = pd.DataFrame(data)
print(df.head())

# 1. Filtrar actores cuyo primer nombre empieza con "A"
df_filtered = df[df['first_name'].str.startswith('A')].copy()  # Aquí creamos una copia explícita
print("\n\n1. Filtrar actores cuyo primer nombre empieza con A:  \n\n", df_filtered)

# 2. Crear una columna 'full_name' concatenando 'first_name' y 'last_name'
df_filtered['full_name'] = df_filtered['first_name'] + ' ' + df_filtered['last_name']
print("\n\n2. Crear una columna 'full_name' concatenando 'first_name' y 'last_name'  \n\n", df_filtered)

# 3. Calcular la longitud total del nombre en una nueva columna 'name_length'
df_filtered['name_length'] = df_filtered['full_name'].apply(len)
print("\n\n3. Calcular la longitud total del nombre en una nueva columna 'name_length'  \n\n", df_filtered)

# 4. Convertir todos los nombres a mayúsculas (tanto 'first_name' como 'last_name')
df_filtered['first_name'] = df_filtered['first_name'].str.upper()
df_filtered['last_name'] = df_filtered['last_name'].str.upper()
print("\n\n4. Convertir todos los nombres a mayúsculas (tanto 'first_name' como 'last_name')  \n\n", df_filtered)

# 5.Filtrar actores cuya longitud del nombre completo sea mayor que 10 caracteres
# Se eliminarán aquellos actores cuyo nombre completo (la concatenación de first_name y last_name) tenga 10 caracteres o menos.
df_filtered = df_filtered.where(df_filtered['full_name'].apply(len)>10)
print("\n\n5.Filtrar actores cuya longitud del nombre completo sea mayor que 10 caracteres)  \n\n", df_filtered)

# 6. Calcular el año de registro de cada actor
# Se extraerá el año del campo last_update para determinar el año en que se registró el actor en la base de datos.
df_filtered['registration_year'] = pd.to_datetime(df_filtered['last_update']).dt.year
print("\n\n6. Calcular el año de registro del actor  \n\n", df_filtered)

# 7. Agrupar por 'last_name' y contar cuántos actores tienen ese apellido
# Esta transformación agrupa a los actores por su apellido (last_name) y cuenta el número total de actores que comparten cada apellido.
df_grouped = df_filtered.groupby('last_name').agg(
    total_actors=('actor_id', 'count')
).reset_index()
print("\n\n7. Agrupar por 'last_name' y contar cuántos actores tienen ese apellido  \n\n", df_grouped)

# 8.Agregar una columna que indique si el actor tiene un nombre único
# Esta transformación agrega una columna que verifica si un actor tiene un nombre completo único en la base de datos. Si el nombre completo (full_name) es único (es decir, no hay otro actor con el mismo nombre completo), la columna indicará "True", de lo contrario, mostrará "False".
#df_filtered['is_unique_name'] = df_filtered.groupby('full_name')['full_name'].transform('count') == 1
df_filtered['is_unique_name'] = df_filtered.groupby('full_name')['full_name'].transform('count') == 1
print("\n\n8.Agregar una columna que indique si el actor tiene un nombre único  \n\n", df_filtered)

# 9. Agregar una columna 'actor_status' que indique si el actor tiene más de 10 años en la base de datos
# Esta transformación crea una columna actor_status que indica el estado del actor en función de su antigüedad en la base de datos. 
# Si el actor_id es mayor a 10, se considera que el actor ha estado más de 10 años en la base de datos y se marca como "Active". Si no, se clasifica como "Inactive".
df_filtered['actor_status'] = df_filtered['actor_id'].apply(lambda x: 'Active' if x > 10 else 'Inactive')
print("\n\n9. Agregar una columna 'actor_status' que indique si el actor tiene más de 10 años en la base de datos  \n\n", df_filtered)

# 10. Eliminar la columna 'last_update' de df_filtered
# Esta transformación elimina la columna last_update del DataFrame df_filtered. 
# Dado que esta columna no se va a utilizar para las transformaciones posteriores, 
# se elimina para simplificar el conjunto de datos y reducir el espacio de almacenamiento, 
# manteniendo solo las columnas relevantes para el análisis y la carga en las tablas finales.
df_filtered = df_filtered.drop(columns=['last_update'])
print("\n\n10. Eliminar la columna 'last_update' de df_filtered  \n\n", df_filtered)

connection.close()

#Cargar los datos 

#Python

# Establecemos nuevamente la conexión con MySQL porque ya la habiamos cerrado
"""connection = mysql.connector.connect(
    host="localhost",         # Cambia según tu configuración
    user="root",              # Cambia según tu configuración
    password="12345678"       # Cambia según tu configuración
)"""

engine = create_engine("mysql+pymysql://root:12345678@localhost/sakila")
# Probar conexión 
connection = engine.connect() 
print("Conexión exitosa a MySQL", type(connection))

#cursor = connection.cursor()

# 1. Crear la base de datos 'sakila_etl' si no existe
create_database_query = "CREATE DATABASE IF NOT EXISTS sakila_etl"
#cursor.execute(create_database_query)
connection.execute(text(create_database_query))


# 2. Seleccionar la base de datos 'sakila_etl' para usarla
#cursor.execute("USE sakila_etl")
connection.execute(text("USE sakila_etl"))

# 3. SQL para crear la tabla 'actor_transformed'
create_actor_transformed_table = """
CREATE TABLE IF NOT EXISTS actor_transformed (
    actor_id INT PRIMARY KEY,              -- Identificador único del actor
    first_name VARCHAR(50),                 -- Nombre del actor
    last_name VARCHAR(50),                  -- Apellido del actor
    full_name VARCHAR(100),                 -- Nombre completo concatenado
    name_length INT,                        -- Longitud del nombre completo
    registration_year INT,                  -- Año de registro (basado en 'last_update')
    is_unique_name BOOLEAN,                 -- Si el nombre es único en la base de datos
    actor_status VARCHAR(10)                -- Estado del actor (Activo/Inactivo)
);
"""

# 4. SQL para crear la tabla 'actor_count_by_last_name'
create_actor_count_by_last_name_table = """
CREATE TABLE IF NOT EXISTS actor_count_by_last_name (
    last_name VARCHAR(50) PRIMARY KEY,      -- Apellido del actor
    total_actors INT                        -- Total de actores con ese apellido
);
"""

# Ejecutar las consultas para crear las tablas
#cursor.execute(create_actor_transformed_table)
connection.execute(text(create_actor_transformed_table))

#cursor.execute(create_actor_count_by_last_name_table)
connection.execute(text(create_actor_count_by_last_name_table))


# Confirmar los cambios y cerrar la conexión
connection.commit()
#cursor.close()
connection.close()

print("Base de datos 'sakila_etl' y tablas 'actor_transformed' y 'actor_count_by_last_name' creadas exitosamente en MySQL")

# Paso 4: Carga de datos con pandas y SQLAlchemy:

# Conexión a la nueva base de datos (sakila_etl)
engine = create_engine("mysql+pymysql://root:12345678@localhost/sakila_etl")
#engine = create_engine('mysql+mysqlconnector://root:12345678@localhost/sakila_etl')


# Cargar los datos filtrados y transformados en la tabla 'actor_transformed'

# Aquí estamos eligiendo las columnas que definimos en el proceso de transformación
df_filtered[['actor_id', 'first_name', 'last_name', 'full_name', 'name_length', 'registration_year', 'is_unique_name', 'actor_status']].to_sql(
    'actor_transformed',    # Nombre de la tabla en MySQL
    con=engine.connect(),             # Conexión al motor de MySQL
    if_exists='replace',    # Si la tabla ya existe, la reemplazamos
    index=False            # No guardamos el índice de pandas como columna
)

# Cargar los datos agrupados (total de actores por apellido) en la tabla 'actor_count_by_last_name'
df_grouped.to_sql(
    'actor_count_by_last_name',   # Nombre de la tabla en MySQL
    con=engine.connect(),                   # Conexión al motor de MySQL
    if_exists='replace',          # Si la tabla ya existe, la reemplazamos
    index=False                   # No guardamos el índice de pandas como columna
)

# Confirmar que los datos se han cargado correctamente
print("Datos cargados exitosamente en la base de datos 'sakila_etl'.")


# Paso 5: Verificación en MySQL 

# Establecemos nuevamente la conexión con MySQL
"""connection = mysql.connector.connect(
    host="localhost",         # Cambia según tu configuración
    user="root",              # Cambia según tu configuración
    password="12345678",   # Cambia según tu configuración
    database="sakila_etl"     # Usamos la base de datos 'sakila_etl'
)"""

engine = create_engine("mysql+pymysql://root:12345678@localhost/sakila_etl")

conexion = engine.connect()

# 1. Ejecutar consulta para seleccionar todos los registros de 'actor_transformed'
select_actor_transformed = "SELECT * FROM actor_transformed"
result = conexion.execute(text(select_actor_transformed))
actor_transformed_records = pd.DataFrame(result)

# Mostrar resultados de 'actor_transformed'
print("Registros de 'actor_transformed':")
for record in actor_transformed_records:
    print(record)

# 2. Ejecutar consulta para seleccionar todos los registros de 'actor_count_by_last_name'
select_actor_count_by_last_name = "SELECT * FROM actor_count_by_last_name"
result2 = conexion.execute(text(select_actor_count_by_last_name))
actor_count_by_last_name_records = pd.DataFrame(result)

# Mostrar resultados de 'actor_count_by_last_name'
print("\nRegistros de 'actor_count_by_last_name':")
for record in actor_count_by_last_name_records:
    print(record)

# Cerrar el cursor y la conexión
conexion.close()
#connection.close()

print("\nConsultas ejecutadas y resultados mostrados exitosamente.")

Iris Aguilar

student•

*Corrección: Base de Datos

Layla Scheli

teacher•

Gracias Iris, siempre pasa, que bueno que lo lograste :)

Transformaciones de Datos ETL con Python y MySQL

Fundamentos de ETL con Python

Procesos ETL: Extracción, Transformación y Carga de Datos

Manipulación de Datos con Python: De CSV a ETL Completo

Fuentes de Datos: Archivos, Bases de Datos y APIs en Análisis de Ventas

Instalación de Anaconda y Jupyter para Procesamiento de Datos

Técnicas Efectivas de Transformación de Datos

Lectura y Exploración de Archivos CSV con Pandas en Python

Manejo de Archivos Excel con Pandas para Procesos ETL

Ingesta de Datos desde APIs con Python

Conexión y consulta de bases de datos en Python con SQLite3 y SQLAlchemy

Series y DataFrames en Pandas: Diferencias y Usos Prácticos

Métricas Clave en el Perfilado de Datos para Análisis de Calidad

Limpieza de Datos en Python con Pandas

Filtrado, Selección y Transformación de Datos con Pandas

Agrupación y Resumen de Datos con Pandas

Manipulación Avanzada de Datos con Python y Pandas

Carga de Datos y Proyecto Final

Exportación de DataFrames a CSV con Pandas paso a paso

Gestión de Datos: Carga Completa vs. Carga Incremental en Python

Partición de Datos en Archivos CSV con Pandas

Carga Incremental de Datos en Archivos Excel con Python

Importación de la Base de Datos Sakila en MySQL Workbench

Transformaciones de Datos ETL con Python y MySQL

Transformaciones de Datos con Python y SQL

Buenas prácticas y errores comunes en procesos ETL

Carga y Validación de Datos en Sakila ETL con Python y SQL

Fundamentos de ETL con Python: Extracción, Transformación y Carga de Datos