Как создать легко воспроизводимый DS проект

Аналитику или исследователю данных приходится разрабатывать множество алгоритмов по обработке и анализу различных данных. Большинство алгоритмов разрабатываются для многоразового использования, а значит, код либо запускается разработчиком с определенной периодичностью, либо код передается другим пользователям для обработки своих данных. При этом алгоритмы имеют множество параметров и зависимостей, которые необходимо индивидуально настраивать под определенные данные.

Для того чтобы сделать процесс развертывания, использования и доработки алгоритма интуитивно понятным воспользуемся инструментом Kedro. Основная концепция kedro заключается в модульной структуре, где весь цикл работы с данными формируется из отдельных блоков в единый рабочий процесс. Проект на kedro имеет следующую структуру:

Где директория conf содержит конфигурационные файлы для подключения к внешним источникам, учетные данные, настройки ml-алгоритма и прочие параметры. data необходима для хранения данных на различных этапах обработки, docs предназначена для документации проекта, а logs для хранения журнала событий работы алгоритмов. Notebooks – предназначен для хранения jupyter-блокнотов и src, в которой размещается основной код для формирования pipeline проекта.

Для примера, возьмем хорошо известную задачу: предсказание выживших при кораблекрушении Титаника. Полный код, представленный в статье, можно найти в репозитории.

В первую очередь инициализируем git, создадим виртуальное окружение и установим kedro. Далее создадим файл config.yml со следующим содержимым:

output_dir: .
project_name: titanic
repo_name: titanic
python_package: titanic

А затем выполним команду kedro new —config config.yml для формирования оболочки будущего проекта.

Чтобы использовать функционал kedro необходимо установить зависимости с помощью команды: kedro install. Следующий шаг – регистрация входных данных, которые будут анализироваться. Для этого поместим данные в titanic\data\01_raw. Все данные, которые будут загружаться kedro, необходимо указать в конфигурационном файле titanic\conf\base\catalog.yml следующим образом:

input_data:
    type: pandas.CSVDataSet
    filepath: data/01_raw/train.csv

Где,

input_data – имя объекта данных, по которому будем обращаться в kedro;

type – модуль kedro, с помощью которого будем загружать данные;

filepath – расположение исходных данных.

Kedro поддерживает множество типов данных: csv, xlsx, json, parquet, sql-query и т.д. Следующий этап – настройка pipeline, который будет выполнять весь цикл работы алгоритма, от предобработки данных до оценки качества модели. Pipeline в Kedro выполняется с помощью nodes – блока модулей, которые выполняются последовательно, а полученные данные передаются от одного модуля к другому. Для начала напишем модуль предобработки данных. В директории src\titanic\pipelines создадим папку с именем data_processing, в которую поместим файл nodes.py со следующим содержимым:

import pandas as pd

def cat_to_num(arr):
    return {i:c for c, i in enumerate(arr)}
    
def preprocess_dataset(df: pd.DataFrame) -> pd.DataFrame:
    df = df.dropna(subset=["Embarked"])
    d_sex = cat_to_num(df["Sex"].unique())
    d_embarked = cat_to_num(df["Embarked"].unique()) 
    df["Sex"] = df["Sex"].map(d_sex)
    df["Embarked"]= df["Embarked"].map(d_embarked)
    return df

Для примера преобразовали 2 столбца («Sex» и «Embarked») к числовому типу и удалили строки, с пропущенными значениями по столбцу Embarked. Далее для инициализации обработки данных в pipeline нам необходимо создать файл pipeline.py рядом с nodes.py в директории src\titanic\pipelines\data_processing со следующим содержимым:

from kedro.pipeline import Pipeline, node
from .nodes import preprocess_dataset

def create_pipeline(**kwargs):
    return Pipeline(
                    [node(
                          func=preprocess_dataset,
                          inputs="input_data",
                          outputs="preprocessed_input_data",
                          name="preprocessing_input_data_node"
                          )
                    ]
                   )

Где,

func – метод для обработки входных данных, который реализован в nodes.py;

inputs – входные данные, имя указано в conf\base\catalog.yml;

outputs – задаем имя для обработанных данных;

name – задаем имя для модуля в pipeline. А также инициализируем pipeline в проектном пространстве, для этого создадим файл __init__.py в директории src\titanic\pipelines\data_processing\, в котором пропишем метод create_pipeline:

from .pipeline import create_pipeline

Заключительный момент создания отдельного модуля pipeline – регистрация в файле src\titanic\ pipeline_registry.py:

from typing import Dict
from kedro.pipeline import Pipeline
from titanic.pipelines import data_processing as dp

def register_pipelines() -> Dict[str, Pipeline]:
    
    data_processing_pipeline = dp.create_pipeline()
    
    return {"__default__": data_processing_pipeline,
            "dp": data_processing_pipeline}

Запустим проверку, чтобы убедиться в правильном составлении модуля обработки входных данных, для этого в терминале запустим команду, что запустит весь pipeline:

kedro run

или запустим отдельный модуль, указав имя через аргумент –node:

kedro run --node=preprocessing_input_data_node

В результате должны получить подобный вывод:

Обработка выполнена, однако результат обработки хранится в памяти (MemoryDataSet) до завершения pipeline. Для того чтобы зафиксировать промежуточный результат необходимо дополнить файл titanic\conf\base\catalog.yml:

………………………….
preprocessed_input_data:
    type: pandas.CSVDataSet
    filepath: data/02_intermediate/preprocessed_input_data.csv

Таким образом, результат метода preprocess_dataset будет сохранен в csv файл, в директорию data/02_intermediate/, к которому можно будет обратиться в любой момент. Далее приступим к DS составляющей pipeline. Для начала необходимо установить пакеты по работе с ml-алгоритмами, для этого отредактируем файл titanic\src\requirements.in:

………
scikit-learn==0.23.1

И вызовем в терминале команду:

kedro install –build-reqs

По подобию с обработкой данных, создадим папку data_science в директории src\titanic\pipelines\ и добавим файл nodes.py со следующим содержимым:

import logging
from typing import Dict, Tuple

import pandas as pd
import numpy as np

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn import metrics

def get_score(name_metric: str, y_true: pd.Series, y_pred: (np.ndarray, np.array)) -> float:
    ## проверка на правильность названия метрики
    name_metrics = [i for i in dir(metrics) if callable(getattr(metrics, i)) and not i.startswith("__")]
    if name_metric not in name_metrics:
        print("Invalid metric name")
        return np.nan
    score = getattr(metrics, name_metric)(y_true, y_pred)
    return round(score, 2)

def split_data(df: pd.DataFrame, parameters: Dict) -> Tuple:
    X = df[parameters["features"]]
    y = df["Survived"]
    X_train, X_test, y_train, y_test = train_test_split(X, 
                                                        y,
                                                        test_size=parameters["test_size"],
                                                        random_state=parameters["random_state"])
    return X_train, X_test, y_train, y_test
    
def train_model(X_train: pd.DataFrame, y_train: pd.Series, parameters: Dict) -> RandomForestClassifier:
    rndm_forest = RandomForestClassifier(n_estimators=parameters["n_estimators"])
    rndm_forest.fit(X_train, y_train)
    return rndm_forest

def evaluate_model(rndm_forest: RandomForestClassifier, X_test: pd.DataFrame, y_test: pd.Series, parameters: Dict):
    y_pred = rndm_forest.predict(X_test)
    score = get_score(parameters["metric"], y_test, y_pred)
    logger = logging.getLogger(__name__)
    logger.info(f"Model has a value {score} for {parameters['metric']}")

Метод split_data разбивает обработанные данные на обучающую и тестовую выборку, train_model – обучение модели, evaluate_model -проверяет качество модели, метод get_score возвращает значение по заданной метрике.

Как вы уже заметили, на вход методов передается аргумент parameters, который мы не импортируем и не извлекаем из других методов. Данный объект будет подтягиваться из конфигурационного файла titanic\conf\base\parameters.yml, из которого будем брать необходимые значения для настройки гиперпараметров модели, размера тестовой выборки и т.д. Метод get_score реализован для примера, чтобы показать возможность настройки любых параметров, которые будут полезны при настройке вашего проекта. Таким образом вам не нужно искать параметры в исходном коде, нужно лишь подправить файл parameters.yml и вызвать команду kedro run, для обучения модели с новыми параметрами. Внесем необходимые значения в файл parameters.yml:

test_size: 0.2
random_state: 17
n_estimators: 100
metric: "accuracy_score"
features:
    - Pclass
    - Sex
    - SibSp
    - Parch
    - Embarked

Параметр features (используем в методе split_data) необходим для отбора только тех признаков, которые будем использовать для обучения и предсказания, также можно указать и целевой признак, если данные берутся из различных источников. Далее повторяем операции сохранения данных, и инициализации pipeline. Для начала дополним файл titanic\conf\base\catalog.yml для сохранения обученной модели:

………….
classificator:
    type: pickle.PickleDataSet
    filepath: data/06_models/classificator.pickle
    versioned: true

Отличием от предыдущих способов сохранения данных является аргумент versioned, если укажем значение true, модель будет сохраняться каждый раз при запуске pipeline. Далее создадим data_science модуль для pipeline, для этого поместим в директорию src\titanic\pipelines\data_science файл pipeline.py со следующим содержимым:

from kedro.pipeline import Pipeline, node
from .nodes import split_data, train_model, evaluate_model

def create_pipeline(**kwargs):
    return Pipeline([
                      node(
                            func=split_data,
                            inputs=["preprocessed_input_data", "parameters"],
                            outputs=["X_train", "X_test", "y_train", "y_test"]
                            name="split_data_node"
                           ),
                      node(
                            func=train_model,
                            inputs=["X_train", "y_train", "parameters"],
                            outputs="classificator",
                            name="train_model_node",
                          ),
                      node(
                            func=evaluate_model,
                            inputs=["classificator", "X_test", "y_test", "parameters"],
                            outputs=None,
                            name="evaluate_model_node"
                           )
                    ])

Данный модуль состоит из 3-х функций, где использование входных данных зависит от выполнения предыдущих функций. Не забываем подавать на вход функций объект parameters, для доступа к параметрам, влияющих на обучающую способность модели.

Обратите внимание, каким образом функции обращаются к данным. В kedro указывается не физическое расположение данных, а используется имя объекта, которое хранится либо в памяти (MemoryDataSet — пока выполняется pipeline), либо ссылается на уже существующий объект, указанный в titanic\conf\base\catalog.yml. Также инициализируем data_science модуль в проектном пространстве, для этого создадим файл __init__.py в директории src\titanic\pipelines\data_science, в котором укажем следующую строку:

from .pipeline import create_pipeline

Также зарегистрируем data_science модуль в общий pipeline проекта, для этого редактируем файл \src\titanic\ pipeline_registry.py:

from typing import Dict
from kedro.pipeline import Pipeline
from titanic.pipelines import data_processing as dp
from titanic.pipelines import data_science as ds

def register_pipelines() -> Dict[str, Pipeline]:

    data_processing_pipeline = dp.create_pipeline()
    data_science_pipeline = ds.create_pipeline()
    
    return {"__default__": data_processing_pipeline + data_science_pipeline,
            "dp": data_processing_pipeline,
            "ds": data_science_pipeline
           }

В данном случае знак «+» для двух pipelines используется для того, чтобы kedro понимал очередность выполнения модулей. Т.к. предобработку данных уже запускали раннее, можно запустить только ds часть с помощью аргумента –pipeline:

kedro run –pipeline=ds

В результате должны получить подобный вывод:

В консоли logger выводит сообщение о том, что точность модели на валидационных данных составляет 0.78 для метрики accuracy_score. По завершению pipeline, обученная модель появится в директории data/06_models/.

На этом закончим формирование pipeline. Создание модуля предсказания на тестовых данных, с лучшей моделью оставим для домашнего задания)))

Код в статье может показаться немного запутанным. Однако у kedro подробная и последовательная документация, которая ответит на любые вопросы по формированию проекта.

Подведем итоги:

Kedro – очень удобен для командной разработки, т.к. имеет определенную структуру, и не требуется поиск определенного параметра для настройки алгоритма в папке «temp» с названием файла «my_functions.py».

Для того чтобы прогнать новые данные, требуется лишь поместить данные в директорию data\01_raw и запустить команду в терминале — kedro run. Но главное – это наличие множества интересных плагинов для kedro, которые упрощают работу, как разработчика, так и исследователя данных. В следующей части статьи уделим внимание возможностям kedro, для визуального представления работы Вашего проекта.