Magia e Músculos: ETL com Mage e DuckDB com dados do meu treino de powerlifting

Acesse o pipeline completo aqui

Mage

No ultimo post que fiz, falei um pouco sobre um painel que construí utilizando Python e o Looker Studio, para visualizar os dados do meu treino de powerlifting. Nesse post vou passar pelo passo a passo de construir um pipeline de ETL (extração, transformação e carga), com os mesmos dataset do meu treino.

Para a construção desse pipeline vamos usar o Mage para orquestração, Python para carga e tratamento dos dados e por fim vamos exportar os dados para uma base do DuckDB.

Para executar o Mage vamos usar a imagem oficial do framework:

docker pull mageai/mageai:latest

O pipeline que vamos construir ficará assim:

Extração

A extração será simples, vamos ler um arquivo CSV usando a biblioteca pandas e utilizar um DataFrame para realizar as etapas seguintes. Usando o bloco Data loader conseguimos um template para ler o arquivo CSV e carregar para um DataFrame do Pandas.

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'

Transformação

Nesta etapa utilizamos o bloco Transformer que possui diversos templates para tratamento de dados, porem para o nosso caso vamos realizar uma transformação que não possui um template definido.

A transformação que vamos realizar consiste no mapeamento da coluna Exercise Name para criação de uma nova coluna chamada Body part onde vamos poder identificar a que parte do corpo pertence cada um dos exercícios.

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'

Uma característica interessante do Mage é que podemos visualizar de maneira bem fácil as mudanças que estamos fazendo, dentro do bloco Transformer na opção Add chart podemos gerar um gráfico de pizza que nos mostram as porcentagens da coluna Body part.

Carga

Por fim, vamos carregar os dados com a nova coluna para nossa base no DuckDB, que por padrão já esta incluída na mesma imagem do Docker em que o Mage esta sendo executado.

Usando o bloco Data exporter vamos selecionar o template do SQL para exportar os dados, e com uma única linha realizamos a inserção dos dados no DuckDB

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 }};

Podemos visualizar a tabela, usando a opção Add chart do bloco Data exporter, e slecionando a opção Table.

Conclusão

Mage é um orquestrador bastante poderoso e flexível, que permite a integração de diversas tecnologias voltas a ETL, nesse post passamos apenas por uma rápida introdução do que é possível fazer com essa ferramenta, em posts futuros pretendo explorar mais a funcionalidades da ferramenta.