Puedes acceder al pipeline completo aquí

mago

En mi última publicación, escribí sobre un panel que construí usando Python y Looker Studio, para visualizar mis datos de entrenamiento de levantamiento de pesas. En esta publicación, lo guiaré paso a paso de una canalización ETL (Extracción, Transformación, Carga), utilizando el mismo conjunto de datos.

Para construir el pipeline usaremos Mage para organizar el pipeline y Python para transformar y cargar los datos, como último paso exportaremos los datos transformados a una base de datos DuckDB.

Para ejecutar Mage, vamos a usar la imagen oficial de Docker:

docker pull mageai/mageai:latest

La canalización se verá así:

Extracto

La extracción será sencilla, solo tenemos que leer un archivo csv y crear un dataframe de pandas con él, para poder continuar con los siguientes pasos. Usando el bloque Cargador de datos, ya tenemos una plantilla con la que trabajar, solo recuerde configurar el parámetro "sep" en la función read_csv(), para que los datos se carguen correctamente.

from mage_ai.io.file import FileIO
import pandas as pd

if 'data_loader' not in globals():

    from mage_ai.data_preparation.decorators import data_loader

if 'test' not in globals():

    from mage_ai.data_preparation.decorators import test

@data_loader
def load_data_from_file(*args, **kwargs):

    filepath = 'default_repo/data_strong.csv'
    df = pd.read_csv(filepath, sep=';')  

    return df

@test
def test_output(output, *args) -> None:
    assert output is not None, 'The output is undefined'`

Transformar

En este paso, usando el bloque Transformador, que tiene muchas plantillas para elegir, seleccionaremos una plantilla personalizada.

La transformación que tenemos que hacer es básicamente el mapeo de la columna Nombre del ejercicio, para que podamos identificar qué parte del cuerpo corresponde al ejercicio específico.

import pandas as pd

if 'transformer' not in globals():

    from mage_ai.data_preparation.decorators import transformer

if 'test' not in globals():

    from mage_ai.data_preparation.decorators import test

body_part = {'Squat (Barbell)': 'Pernas',

    'Bench Press (Barbell)': 'Peitoral',

    'Deadlift (Barbell)': 'Costas',

    'Triceps Pushdown (Cable - Straight Bar)': 'Bracos',

    'Bent Over Row (Barbell)': 'Costas',

    'Leg Press': 'Pernas',

    'Overhead Press (Barbell)': 'Ombros',

    'Romanian Deadlift (Barbell)': 'Costas',

    'Lat Pulldown (Machine)': 'Costas',

    'Bench Press (Dumbbell)': 'Peitoral',

    'Skullcrusher (Dumbbell)': 'Bracos',

    'Lying Leg Curl (Machine)': 'Pernas',

    'Hammer Curl (Dumbbell)': 'Bracos',

    'Overhead Press (Dumbbell)': 'Ombros',

    'Lateral Raise (Dumbbell)': 'Ombros',

    'Chest Press (Machine)': 'Peitoral',

    'Incline Bench Press (Barbell)': 'Peitoral',

    'Hip Thrust (Barbell)': 'Pernas',

    'Agachamento Pausado ': 'Pernas',

    'Larsen Press': 'Peitoral',

    'Triceps Dip': 'Bracos',

    'Farmers March ': 'Abdomen',

    'Lat Pulldown (Cable)': 'Costas',

    'Face Pull (Cable)': 'Ombros',

    'Stiff Leg Deadlift (Barbell)': 'Pernas',

    'Bulgarian Split Squat': 'Pernas',

    'Front Squat (Barbell)': 'Pernas',

    'Incline Bench Press (Dumbbell)': 'Peitoral',

    'Reverse Fly (Dumbbell)': 'Ombros',

    'Push Press': 'Ombros',

    'Good Morning (Barbell)': 'Costas',

    'Leg Extension (Machine)': 'Pernas',

    'Standing Calf Raise (Smith Machine)': 'Pernas',

    'Skullcrusher (Barbell)': 'Bracos',

    'Strict Military Press (Barbell)': 'Ombros',

    'Seated Leg Curl (Machine)': 'Pernas',

    'Bench Press - Close Grip (Barbell)': 'Peitoral',

    'Hip Adductor (Machine)': 'Pernas',

    'Deficit Deadlift (Barbell)': 'Pernas',

    'Sumo Deadlift (Barbell)': 'Costas',

    'Box Squat (Barbell)': 'Pernas',

    'Seated Row (Cable)': 'Costas',

    'Bicep Curl (Dumbbell)': 'Bracos',

    'Spotto Press': 'Peitoral',

    'Incline Chest Fly (Dumbbell)': 'Peitoral',

    'Incline Row (Dumbbell)': 'Costas'}


@transformer
def transform(data, *args, **kwargs):
    strong_data = data[['Date', 'Workout Name', 'Exercise Name', 'Weight', 'Reps',    'Workout Duration']]
    strong_data['Body part'] = strong_data['Exercise Name'].map(body_part)

    return strong_data

@test
def test_output(output, *args) -> None:
    assert output is not None, 'The output is undefined'

Una característica interesante de Mage es que podemos visualizar los cambios que estamos realizando dentro del bloque Tranformer, usando la opción Agregar gráfico es posible generar un gráfico circular usando la columna Parte del cuerpo.

Carga

Ahora es el momento de cargar los datos en DuckDB. En la imagen de la ventana acoplable ya tenemos DuckDB, por lo que solo tenemos que incluir otro bloque en nuestra canalización. Incluyamos el bloque Exportador de datos, con una plantilla SQL para que podamos crear una tabla e insertar los datos.

CREATE OR REPLACE TABLE powerlifting 
(
    _date DATE,
    workout_name STRING,
    exercise_name STRING,
    weight STRING,
    reps STRING,
    workout_duration STRING,
    body_part STRING
);

INSERT INTO powerlifting SELECT * FROM {{ df_1 }};

Conclusión

Mage es una herramienta poderosa para orquestar canalizaciones y proporciona un conjunto completo de plantillas para desarrollar tareas específicas que involucran ETL. En esta publicación, tuvimos una breve explicación sobre cómo comenzar a usar Mage para crear canales de datos. En publicaciones futuras exploraremos más sobre este marco tan sorprendente.