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