Вы можете получить доступ ко всему конвейеру здесь

Маг

В моем последнем посте я писал о панели мониторинга, которую я создал с помощью Python и Looker Studio для визуализации данных о тренировках по пауэрлифтингу. В этом посте я шаг за шагом покажу вам работу конвейера ETL (извлечение, преобразование, загрузка) с использованием того же набора данных.

Для построения конвейера мы будем использовать Mage для оркестровки конвейера и Python для преобразования и загрузки данных. На последнем этапе мы экспортируем преобразованные данные в базу данных DuckDB.

Чтобы запустить Mage, мы собираемся использовать официальный образ докера:

docker pull mageai/mageai:latest

Конвейер будет выглядеть так:

Извлекать

Извлечение будет простым, нам просто нужно прочитать файл csv и создать с его помощью фрейм данных pandas, чтобы мы могли перейти к следующим шагам. Используя блок загрузчика данных, у нас уже есть шаблон для работы, просто не забудьте установить параметр «sep» в функцию read_csv(), чтобы данные загружались правильно.

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

Трансформировать

На этом этапе, используя блок «Трансформер», в котором есть множество шаблонов на выбор, мы выберем собственный шаблон.

Преобразование, которое нам нужно выполнить, по сути, представляет собой сопоставление столбца «Название упражнения», чтобы мы могли определить, какая часть тела соответствует конкретному упражнению.

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, используя опцию «Добавить диаграмму», выбрав возможность создания круговой диаграммы с использованием части тела столбца.

Нагрузка

Пришло время загрузить данные в DuckDB. В образе докера у нас уже есть DuckDB, поэтому нам просто нужно включить еще один блок в наш конвейер. Давайте добавим блок Data Exporter с шаблоном 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 для построения конвейеров данных. В следующем посте мы собираемся подробнее изучить эту замечательную структуру.