Vous pouvez accéder au pipeline complet ici

mage

Dans mon dernier article, j'ai parlé d'un tableau de bord que j'ai créé à l'aide de Python et Looker Studio, pour visualiser mes données d'entraînement de dynamophilie. Dans cet article, je vais vous guider étape par étape d'un pipeline ETL (Extract, Transform, Load), en utilisant le même ensemble de données.

Pour construire le pipeline, nous utiliserons Mage pour orchestrer le pipeline et Python pour transformer et charger les données, comme dernière étape, nous exporterons les données transformées dans une base de données DuckDB.

Pour exécuter Mage, nous allons utiliser l'image docker officielle :

docker pull mageai/mageai:latest

Le pipeline ressemblera à ceci :

Extrait

L'extraction sera simple, il suffit de lire un fichier csv et de créer un dataframe pandas avec, afin de pouvoir passer aux étapes suivantes. En utilisant le bloc Data Loader, nous avons déjà un modèle avec lequel travailler, n'oubliez pas de définir le paramètre "sep" dans la fonction read_csv(), afin que les données soient chargées correctement.

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'`

Transformer

Dans cette étape, en utilisant le bloc Transformer, qui propose de nombreux modèles, nous sélectionnerons un modèle personnalisé.

La transformation que nous devons effectuer est essentiellement le mappage de la colonne Nom de l'exercice, afin que nous puissions identifier quelle partie du corps correspond à l'exercice spécifique.

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'

Une fonctionnalité intéressante de Mage est que nous pouvons visualiser les modifications que nous apportons à l'intérieur du bloc Transformer, en utilisant l'option Ajouter un graphique, il est possible de générer un graphique circulaire en utilisant la colonne Partie du corps.

Charger

Il est maintenant temps de charger les données dans DuckDB. Dans l'image Docker, nous avons déjà DuckDB, il nous suffit donc d'inclure un autre bloc dans notre pipeline. Incluons le bloc Data Exporter, avec un modèle SQL afin que nous puissions créer une table et insérer les données.

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

Conclusion

Mage est un outil puissant pour orchestrer des pipelines, fournir un ensemble complet de modèles pour développer des tâches spécifiques impliquant ETL. Dans cet article, nous avons eu une explication approfondie sur la façon de commencer à utiliser Mage pour créer des pipelines de données. Dans un prochain article, nous allons en savoir plus sur ce cadre amusant.