Você pode acessar o pipeline completo aqui

Mago

Em meu último post, escrevi sobre um painel que construí usando Python e Looker Studio, para visualizar meus dados de treinamento de powrlifting. Neste post vou orientar você passo a passo de um pipeline ETL (Extrair, Transformar, Carregar), usando o mesmo conjunto de dados.

Para construir o pipeline usaremos Mage para orquestrar o pipeline e Python para transformar e carregar os dados, como último passo exportaremos os dados transformados para um banco de dados DuckDB.

Para executar o Mage, vamos usar a imagem oficial do docker:

docker pull mageai/mageai:latest

O pipeline ficará assim:

Extrair

A extração será simples, basta ler um arquivo csv e criar um dataframe do pandas com ele, para podermos prosseguir para os próximos passos. Utilizando o bloco Data loader, já temos um tamplate para trabalhar, basta lembrar de definir o parâmetro "sep" na função read_csv(), para que os dados sejam carregados corretamente.

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

Nesta etapa, usando o bloco Transformer, que possui vários modelos para escolher, selecionaremos um modelo personalizado.

A transformação que temos que fazer é basicamente o mapeamento da coluna Nome do Exercício, para que possamos identificar qual parte do corpo corresponde ao exercício 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'

Uma característica interessante do Mage é que podemos visualizar as mudanças que estamos fazendo dentro do bloco Tranformer, usando o gráfico Add, optando por ser possível gerar um gráfico de pizza usando a coluna Body part.

Carregar

Agora é hora de carregar os dados no DuckDB. Na imagem docker já temos o DuckDB, então só precisamos incluir outro bloco em nosso pipeline. Vamos incluir o bloco Data Exporter, com um template SQL para que possamos criar uma tabela e inserir os dados.

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

Conclusão

Mage é uma ferramenta poderosa para orquestrar pipelines, fornece um conjunto completo de templates para o desenvolvimento de tarefas específicas envolvendo ETL. Neste post tivemos uma explicação resumida sobre como começar a usar o Mage para construir pipelines de dados. Em uma postagem futura, exploraremos mais sobre essa estrutura divertida.