Transformaciones de Datos ETL con Python y MySQL

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

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

Enrique Alexis Lopez Araujo

student•

Para conectar a MySQL y realizar las transformaciones mencionadas en clase, puedes utilizar el siguiente código en Python:

import mysql.connector
import pandas as pd

# Conexión a la base de datos
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='tu_password',
    database='Shakira'
)

# Creación del cursor
cursor = conn.cursor()

# Consulta para extraer datos
cursor.execute("SELECT * FROM actor")
resultados = cursor.fetchall()

# Guardar resultados en un DataFrame
columnas = [desc[0] for desc in cursor.description]
df = pd.DataFrame(resultados, columns=columnas)

# Realizar transformaciones
df_filtered = df[df['first_name'].str.startswith('A')]
df['full_name'] = df['first_name'] + ' ' + df['last_name']
df['name_length'] = df['full_name'].str.len()
df['full_name'] = df['full_name'].str.upper()
df = df[df['name_length'] > 10]
# Agregar las demás transformaciones según se indique...

# Cerrar la conexión
cursor.close()
conn.close()

Asegúrate de reemplazar tu_password con la contraseña correcta. Este código realiza la conexión y parte de las transformaciones que se discutieron en clase.

Daniel Guerrero

student•

Sakila me funciona mejor, ;)

sl2

Enrique Alexis Lopez Araujo

student•

Lo más importante de la clase es entender el proceso de extracción, transformación y carga (ETL) de datos, utilizando Python y MySQL. Un data engineer debe dominar:

Conexiones a bases de datos.
Realización de consultas SQL para extraer datos.
Aplicación de transformaciones a los datos, como filtrado, creación de columnas, y agrupaciones.
Carga de los datos transformados en nuevas bases de datos.

Estos fundamentos son esenciales para construir pipelines de datos eficientes y robustos.

Layla Scheli

teacher•

Gracias Enrique, buen aporte :)

Mario Alexander Vargas Celis

student•

La planificación y extracción de datos desde MySQL en un proyecto ETL requiere una buena estructura para garantizar que los datos se extraigan de manera eficiente y segura. A continuación, te muestro los pasos esenciales:

1. Planificación de la Extracción

a. Definir los Requerimientos de Datos

¿Qué datos necesitas? Identifica las tablas y columnas requeridas.
¿Con qué frecuencia necesitas los datos? Determina si la extracción será completa (todos los datos) o incremental (solo los datos nuevos o actualizados).
¿Qué filtros aplicarás? Especifica condiciones como rangos de fechas o valores específicos.

b. Evaluar el Volumen de Datos

¿Cuántos registros? Estima el tamaño de los datos.
Optimización: Considera índices en las tablas para acelerar las consultas.

c. Asegurar la Conexión y Seguridad

Verifica las credenciales y permisos de acceso.
Asegúrate de que el usuario de MySQL tenga los privilegios necesarios para realizar consultas.

2. Configuración del Entorno en Python

a. Instalación de Librerías

Asegúrate de tener instaladas las bibliotecas necesarias para conectarte a MySQL y trabajar con datos:

pip install mysql-connector-python pandas

3. Extracción de Datos desde MySQL

a. Conexión a la Base de Datos

Crea una conexión a MySQL desde Python:

import mysql.connector

# Conectar a MySQL conexion = mysql.connector.connect( host="localhost", # Dirección del servidor MySQL user="etl_user", # Usuario con acceso password="tu_password", # Contraseña del usuario database="proyecto_etl" # Base de datos a trabajar )

if conexion.is_connected(): print("Conexión exitosa a la base de datos") conexion.close()

b. Consulta para Extracción de Datos

Usa consultas SQL para extraer los datos que necesitas.

Extracción Completa

import pandas as pd

# Conexión conexion = mysql.connector.connect( host="localhost", user="etl_user", password="tu_password", database="proyecto_etl" )

# Consulta SQL query = "SELECT * FROM ventas;" df = pd.read_sql(query, conexion)

# Mostrar datos print(df.head())

conexion.close()

Extracción Incremental

Extrae datos nuevos o actualizados basándote en una columna como fecha.

query = """ SELECT * FROM ventas WHERE fecha >= '2025-01-01'; """ df = pd.read_sql(query, conexion)

4. Optimización de la Extracción

a. Filtrar los Datos en SQL

Realiza filtros en la consulta SQL para reducir el volumen de datos extraídos.

SELECT producto, cantidad, precio, fecha FROM ventas WHERE cantidad > 5 AND fecha >= '2025-01-01';

b. Trabajar con Lotes

Si el volumen de datos es grande, extrae los datos en partes.

cursor = conexion.cursor()

# Consulta con límite query = "SELECT * FROM ventas LIMIT 1000 OFFSET 0;" cursor.execute(query)

for row in cursor.fetchall(): print(row)

cursor.close()

5. Validación y Registro

Asegúrate de registrar cada extracción:

Validación: Comprueba que no haya datos corruptos o faltantes.
Registro: Guarda un log con información de la extracción: fecha, registros extraídos, tiempo de ejecución.

6. Ejemplo Completo

Combina todo en un flujo básico:

import mysql.connector import pandas as pd

# Configuración de conexión conexion = mysql.connector.connect( host="localhost", user="etl_user", password="tu_password", database="proyecto_etl" )

# Extracción de datos query = "SELECT producto, cantidad, precio, fecha FROM ventas WHERE cantidad > 5;" df = pd.read_sql(query, conexion)

# Mostrar resultados print("Registros extraídos:", len(df)) print(df.head())

# Guardar en un archivo CSV df.to_csv("ventas_filtradas.csv", index=False)

conexion.close()

7. Siguientes Pasos

Transformación: Limpia y prepara los datos.
Carga: Inserta los datos procesados en otro sistema, como un Data Warehouse.

Carlos Orejuela

student•

Python cuenta con la declaración with que simplica la gestión de recursos de manera automática. Esto resulta útil para no tener que cerrar la conexión de una base de datos manualmente o para liberar los recursos de un archivo con el que ya no estamos trabajando.

import os

from dotenv import load_dotenv
import mysql.connector
import pandas as pd

# Load db password
load_dotenv()
psswd = os.getenv("password")

# 1) Conexión con la base de datos.
credentials = {
    "user": "cfol",
    "password": psswd,
    "database": "sakila",
    # Default values:
    "host": "127.0.0.1",
    "port": 3306
}

with mysql.connector.connect(**credentials) as con:
    with con.cursor(buffered=True) as cursor:
        # 1) Seleccionar los actores cuyo nombre empieza con 'A'.
        cursor.execute("SELECT * FROM actor WHERE first_name LIKE 'A%'")
        df_actors = pd.DataFrame(cursor.fetchall(), 
                                 columns=[col[0] for col in cursor.description])

Carlos Orejuela

student•

Para los que usaron Docker para instalar MySQL ([tutorial](https://dev.mysql.com/doc/mysql-installation-excerpt/8.0/en/docker-mysql-getting-started.html)), es importante que previo a la ejecución del contenedor hayan especificado el port mapping:

docker run --name=mysql1 \
  --restart on-failure \
  -d \
  -p 3306:3306 \
  container-registry.oracle.com/mysql/community-server:latest

Luego, cuando se conectan a la consola interactiva de mysql, deberán crear su propio usuario con permisos de administador, especificando como dirección IP la del virtual bridge creado por Docker (esto lo puedes consultar ejecutando el comando ifconfig y buscando una sección que sea docker0).

CREATE USER 'docker_user'@'172.17.%' IDENTIFIED BY 'secure_password';
GRANT ALL PRIVILEGES ON *.* TO 'docker_user'@'172.17.%';
FLUSH PRIVILEGES;

Finalmente, cuando quieran inicializar la conexión con la base de datos:

import mysql.connector

credentials = {
  'user': 'docker_user',
  'password': 's3cr3to',
  'database': 'sakila'
}

  with mysql.connector.connect(**credentials) as con:
    # code

Fabrizio Ramírez

student•

Si están utilizando DOCKER para levantar MySQL, también deben añadir el número de puerto.

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

import mysql.connector
import pandas as pd

# Conexión a la base de datos
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='tu_password',
    database='Shakira'
)

# Creación del cursor
cursor = conn.cursor()

# Consulta para extraer datos
cursor.execute("SELECT * FROM actor")
resultados = cursor.fetchall()

# Guardar resultados en un DataFrame
columnas = [desc[0] for desc in cursor.description]
df = pd.DataFrame(resultados, columns=columnas)

# Realizar transformaciones
df_filtered = df[df['first_name'].str.startswith('A')]
df['full_name'] = df['first_name'] + ' ' + df['last_name']
df['name_length'] = df['full_name'].str.len()
df['full_name'] = df['full_name'].str.upper()
df = df[df['name_length'] > 10]
# Agregar las demás transformaciones según se indique...

# Cerrar la conexión
cursor.close()
conn.close()

import os

from dotenv import load_dotenv
import mysql.connector
import pandas as pd

# Load db password
load_dotenv()
psswd = os.getenv("password")

# 1) Conexión con la base de datos.
credentials = {
    "user": "cfol",
    "password": psswd,
    "database": "sakila",
    # Default values:
    "host": "127.0.0.1",
    "port": 3306
}

with mysql.connector.connect(**credentials) as con:
    with con.cursor(buffered=True) as cursor:
        # 1) Seleccionar los actores cuyo nombre empieza con 'A'.
        cursor.execute("SELECT * FROM actor WHERE first_name LIKE 'A%'")
        df_actors = pd.DataFrame(cursor.fetchall(), 
                                 columns=[col[0] for col in cursor.description])

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