您可以在这里访问完整的管道
法师
在我的上一篇文章中，我写了一个使用 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 构建数据管道进行了简短的解释。在以后的文章中，我们将进一步探索这个令人惊奇的框架。
