Sie können hier auf die vollständige Pipeline zugreifen

Magier

In meinem letzten Beitrag habe ich über ein Dashboard geschrieben, das ich mit Python und Looker Studio erstellt habe, um meine Powerlifting-Trainingsdaten zu visualisieren. In diesem Beitrag werde ich Sie Schritt für Schritt durch eine ETL-Pipeline (Extrahieren, Transformieren, Laden) unter Verwendung desselben Datensatzes führen.

Um die Pipeline zu erstellen, verwenden wir Mage zur Orchestrierung der Pipeline und Python zum Transformieren und Laden der Daten. Als letzten Schritt werden wir die transformierten Daten in eine DuckDB-Datenbank exportieren.

Um Mage auszuführen, verwenden wir das offizielle Docker-Image:

docker pull mageai/mageai:latest

Die Pipeline sieht folgendermaßen aus:

Extrakt

Die Extraktion wird einfach sein, wir müssen lediglich eine CSV-Datei lesen und damit einen Pandas-Datenrahmen erstellen, damit wir mit den nächsten Schritten fortfahren können. Mit dem Datenladeblock haben wir bereits eine Vorlage, mit der wir arbeiten können. Denken Sie nur daran, den Parameter „sep“ in der Funktion read_csv( ) festzulegen, damit die Daten korrekt geladen werden.

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

Verwandeln

In diesem Schritt wählen wir mithilfe des Transformer-Blocks, der viele Vorlagen zur Auswahl hat, eine benutzerdefinierte Vorlage aus.

Die Transformation, die wir durchführen müssen, ist im Wesentlichen die Zuordnung der Spalte „Übungsname“, damit wir identifizieren können, welcher Körperteil der spezifischen Übung entspricht.

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'

Eine interessante Funktion von Mage ist, dass wir die Änderungen, die wir innerhalb des Transformer-Blocks vornehmen, visualisieren können. Mit der Option „Diagramm hinzufügen“ ist es möglich, ein Kreisdiagramm mithilfe des Spaltenkörperteils zu erstellen.

Belastung

Jetzt ist es an der Zeit, die Daten in DuckDB zu laden. Im Docker-Image haben wir DuckDB bereits, wir müssen also nur noch einen weiteren Block in unsere Pipeline aufnehmen. Fügen wir den Data Exporter-Block mit einer SQL-Vorlage ein, damit wir eine Tabelle erstellen und die Daten einfügen können.

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

Abschluss

Mage ist ein leistungsstarkes Tool zur Orchestrierung von Pipelines und bietet einen vollständigen Satz an Vorlagen für die Entwicklung spezifischer Aufgaben im Zusammenhang mit ETL. In diesem Beitrag haben wir ausführlich erklärt, wie Sie Mage zum Erstellen von Datenpipelines verwenden können. In zukünftigen Beiträgen werden wir mehr über dieses faszinierende Framework erfahren.