您可以在此处访问完整的管道

法师

在我的上一篇文章中，我写了一个使用 Python 和 Looker Studio 构建的仪表板，用于可视化我的举重训练数据。在这篇文章中，我将使用相同的数据集逐步引导您完成 ETL（提取、转换、加载）管道。

为了构建管道，我们将使用 Mage 来编排管道，并使用 Python 来转换和加载数据，最后一步，我们将转换后的数据导出到 DuckDB 数据库中。

要执行 Mage，我们需要使用官方 docker 镜像：

docker pull mageai/mageai:latest

管道将如下所示：

提炼

提取很简单，我们只需读取 csv 文件并用它创建一个 pandas 数据框，这样我们就可以进行下一步。使用数据加载器块，我们已经有了一个可以使用的模板，只需记住在 read_csv( ) 函数中设置“sep”参数，即可正确加载数据。

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 块，我们将选择一个自定义模板。

我们要做的转换基本上就是Exercise Name列的映射，这样我们就可以识别出哪个身体部位对应于特定的练习。

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'

Mage 的一个有趣的功能是，我们可以使用添加图表来可视化我们在 Tranformer 块内所做的更改，选择可以使用列 Body 部分生成饼图。

加载

现在是时候将数据加载到 DuckDB 了。在 docker 镜像中，我们已经有了 DuckDB，因此我们只需在管道中包含另一个块即可。让我们包含带有 SQL 模板的数据导出器块，以便我们可以创建一个表并插入数据。

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

结论

Mage 是一个功能强大的工具，用于编排管道，提供一整套用于开发涉及 ETL 的特定任务的模板。在这篇文章中，我们对如何开始使用 Mage 构建数据管道进行了简短的解释。在以后的文章中，我们将进一步探讨这个令人惊奇的框架。