Kedro — фреймворк модульного кода в Data Science. С его помощью вы можете создавать проекты по шаблону, настраивать конвейер в YAML, делить его на части, документировать проект — и это далеко не всё. Материалом о работе с Kedro делимся к старту курса по Data Science.

---

Вы когда-нибудь передавали свои данные в список функций и классов, не зная наверняка, каким будет результат? Можно попробовать сохранить данные, а затем проверить их в Jupyter Notebook, чтобы убедиться, что результат соответствует вашим ожиданиям. Этот подход работает, но он трудоёмкий:

def process_data(data):
  pass

def split_data(data):
  pass 

df = process_data(data)
df.to_csv('processed_data.csv')

X_train, X_test, y_train, y_test = split_data(data)

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

def split(data):
  pass 

def train(X_train, y_train):
  pass

def predict(X_test, model):
  pass 

def report(predictions):
  pass

X_train, y_train, X_test = split(data)
model = train(X_train, y_train)
predictions = predict(X_test, model)
report(predictions)

А пока проект растёт, код становится сложнее. Разве не лучше визуализировать связь различных входных и выходных данных? Вот так:

Возможности

Kedro заимствует заимствует концепции лучших практик программной инженерии и применяет их к коду ML. При помощи Kedro можно:

1. Создавать проекты по шаблону.

2. Создать конвейер.

3. Отделять части конвейера.

4. Модуляризовать конвейер.

5. Настраивать данные и параметры через YAML.

6. Анализировать выходные данные узлов Jupyter Notebook.

7. Визуализировать конвейер.

8. Создавать документацию для вашего проекта.

Чтобы установить модуль, введите:

pip install kedro

Начинаем проект 

Создаём проект по шаблону

Вы когда-либо думали, как структурировать свой проект в науке данных, чтобы он был логично и разумно стандартизирован? Создать чётко определённую и стандартную структуру проекта в одной строке кода. Это легко сделать с помощью Kedro. После установки фреймворка введите строку:

kedro new

После ряда вопросов будет создан проект с такой структурой:

Вот 5 основных каталогов:

• conf: конфигурация;

• data: данные;

• docs: документация;

• logs: логи;

• notebooks: блокноты Jupyter Notebook;

• src: код.

Установка зависимостей

До работы с Kedro необходимо установить зависимости из файла src/requirements.txt:

kedro install

Теперь давайте разберёмся, как создать конвейер.

Создание конвейера

Чтобы создать новый конвейер, введите:

kedro pipeline create <NAME>

Проект часто делится на два этапа: обработку данных и обучение модели. Создадим конвейеры data_engineer и data_science :

kedro pipeline create data_engineering
kedro pipeline create data_science

Эти два каталога конвейеров будут созданы в src/project_name:

В каждом конвейере 4 файла:

Узел

Конвейер состоит из нескольких узлов. Каждый узел представляет функцию Python.

from typing import Any, Dict, List
import pandas as pd

def get_classes(data: pd.DataFrame, target_col: str) -> List[str]:
    """Node for getting the classes from the Iris data set."""
    return sorted(data[target_col].unique())

Для каждого узла есть входные и выходные данные:

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

pipeline = Pipeline(
        [
            node(
                get_classes,
                inputs=["raw_data", "params:target_col"],
                ouputs="classes",
                name='get_classes',
            ),
        ]

Визуализация узла get_classes:

Входные и выходные данные каждого узла могут быть заданы как None, строки, список строк или словарь. Обратите внимание, что эти строки — абстрактные имена, а не реальные значения.

Для чего имена? Если мы знаем имена входов и выходов каждой функции, легко получить конкретный вход или выход, вызвав его по имени. Синтаксис определения узла вы найдёте здесь.

Конвейер

Конвейер состоит из списка узлов. Он соединяет выходы одного узла с входами другого узла. Ниже 'classes' (вывод функции get_classes) и 'encoded_data' (вывод функции encode_categorical_columns) используются в качестве входных данных split_data.

from kedro.pipeline import Pipeline, node
from .nodes import get_classes, encode_categorical_columns, split_data


def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                get_classes,
                ["raw_data", "params:target_col"],
                "classes",
                name='get_classes',
            ),
            node(
                encode_categorical_columns,
                ['raw_data', 'params:target_col'],
                'encoded_data',
                name='encode_categorical_columns',
            ),
            node(
                split_data,
                ["encoded_data", "params:test_data_ratio", "classes"],
                dict(
                    train_x="train_x",
                    train_y="train_y",
                    test_x="test_x",
                    test_y="test_y",
                ),
                name="split",
            )
        ]
    )

Каталог данных расположен здесь: conf/base/catalog.yml.

И вот визуализация конвейера:

Регистрация и выполнение

Теперь, когда у нас есть узел и конвейер, давайте зарегистрируем элементы в файле src/project_name/pipeline_registry.py:

"""Project pipelines."""
from typing import Dict

from kedro.pipeline import Pipeline

from iris.pipelines import data_engineering as de
from iris.pipelines import data_science as ds


def register_pipelines() -> Dict[str, Pipeline]:
    """Register the project's pipelines.

    Returns:
        A mapping from a pipeline name to a ``Pipeline`` object.

    """
    data_engineering_pipeline = de.create_pipeline()
    data_science_pipeline = ds.create_pipeline()

    return {
        "de": data_engineering_pipeline,
        "ds": data_science_pipeline,
        "__default__": data_engineering_pipeline + data_science_pipeline,
    }

После регистрации элементов конвейера мы можем запустить их:

kedro run

Если вы хотите запустить конкретный конвейер, добавьте --pipeline=<NAME> к команде kedro run:

kedro run --pipeline=de

Разделение конвейера

Можно запустить только часть конвейера. Есть четыре варианта:

• --from-nodes: выполнение конвейера с определённых узлов;

• --to-nodes: выполнение конвейера до достижения определённых узлов;

• --from-inputs: выполнение конвейера, начиная с узлов с определёнными входными данными;

• --to-outputs: выполнение до определённых выходов.

Команда ниже выполняет конвейер до достижения узла encode_categorical_columns:

kedro run --to-nodes=encode_categorical_columns

Модуляризация

Иногда вы можете захотеть воспользоваться одним и тем же конвейером для разных задач. Kedro позволяет создавать модульные конвейеры, изолированные и переиспользуемые. Вместо отдельных конвейеров «cook lunch pipeline» и «cook dinner pipeline», вы можете написать один — «cook pipeline». А затем превратить «cook pipeline» в «cook meat pipeline» и «cook vegetable pipeline», поменяв входы и выходы «cook pipeline» на новые.

Модульные конвейеры хороши тем, что они переносимы, их легче разрабатывать, тестировать и поддерживать. Инструкции по модуляризации с помощью Kedro можно найти здесь.

Настройка параметров и данных с помощью файла YAML

Параметры

Kedro также позволяет задать параметры для функции с помощью файла YAML. Это очень удобно: вы можете просмотреть все параметры из файла, не копаясь в исходном коде.

def split_data(data: pd.DataFrame, test_data_ratio: float, classes: list) -> Dict[str, Any]:
    """Node for splitting the classical Iris data set into training and test
    sets, each split into features and labels.
    The split ratio parameter is taken from conf/project/parameters.yml.
    The data and the parameters will be loaded and provided to your function
    automatically when the pipeline is executed and it is time to run this node.
    """

    X, y = data.drop(columns=classes), data[classes]
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_data_ratio)

    # When returning many variables, it is a good practice to give them names:
    return dict(
        train_x=X_train,
        train_y=y_train,
        test_x=X_test,
        test_y=y_test,
    )

Чтобы настроить проект с помощью файла конфигурации, начните с размещения параметров, используемых для конвейера data_engineering, в файле conf/base/parameters/data_engineering.yml.

conf/base
├── catalog.yml
├── logging.yml
└── parameters
    ├── data_engineering.yml
    └── data_science.yml

test_data_ratio: 0.2

Теперь получить доступ к параметру из конвейера очень просто. Всё, что нам нужно сделать, — это добавить params: перед именем параметра, к которому мы хотим получить доступ. Ниже params:test_data_ratio используется для доступа к параметру test_data_ratio из конфигурационного файла.

def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                encode_categorical_columns,
                ['raw_data', 'params:target_col'],
                'encoded_data',
                name='encode_categorical_columns',
            ),
        ]
    )

Каталог данных

Kedro позволяет загружать и сохранять данные с помощью Data Catalog. Каталог данных расположен в conf/base/catalog.yml. Это файл конфигурации, который позволяет указать тип данных и место их сохранения. Чтобы сохранить вывод encoded_data из узла encode_categorical_columns, мы можем вставить имя файла encoded_data в conf/base/catalog.yml. В encoded_data указываются тип набора данных и его местоположение.

def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                encode_categorical_columns,
                ['raw_data', 'params:target_col'],
                'encoded_data',
                name='encode_categorical_columns',
            ),
        ]
    )

raw_data:
  type: pandas.CSVDataSet
  filepath: data/01_raw/iris.csv

encoded_data:
  type: pandas.CSVDataSet
  filepath: data/02_intermediate/encoded_data.csv

pandas.CSVDataSet сообщает Kedro, что мы хотим сохранить encoded_data в формате CSV и загрузить их с помощью pandas. Kedro также поддерживает другие типы наборов данных: pickle, parquet, excel, plot, Spark, SQL и др. Здесь весь список.

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

• расположение каждого файла;

• то, как сохраняются и загружаются данные.

Редактируется всего один файл. В следующий раз, когда вы забудете местоположение своих наборов данных, обратитесь к этому файлу.

Загрузка данных и параметров из Jupyter Notebook

Вы когда-нибудь хотели быстро проверить вывод функции из блокнота Jupyter? Как правило, для этого сначала нужно сохранить данные:

# script to process the data

processed_data.to_csv('processed_data.csv')

А затем загрузить в Jupyter Notebook:

В Kedro сохранение и загрузка данных не требуют дополнительного кода. Чтобы начать Jupyter-сессию, введите:

kedro jupyter notebook

После запуска конвейера, который выводит encoded_data, определённый в catalog.yml, данные легко загружается в блокнот с помощью catalog.load.

raw_data:
  type: pandas.CSVDataSet
  filepath: data/01_raw/iris.csv

encoded_data:
  type: pandas.CSVDataSet
  filepath: data/02_intermediate/encoded_data.csv

Если вы хотите указать, как сохранить и загрузить вывод encoded_data, добавьте load_args и save_args в encoded_data.

encoded_data:
  type: pandas.CSVDataSet
  filepath: data/02_intermediate/encoded_data.csv    
  load_args:
    sep: ','
  save_args:
    index: False

Обратите внимание, что приведённая выше конфигурация будет эквивалентна:

import pandas as pd 

df = # Code to produce the data

# Save data
df.to_csv('encoded_data.csv', index=False)

# Load data
df = pd.read_csv('encoded_data.csv', sep=",")

Загрузка параметров

Все параметры внутри src/base также может быть легко загружена с помощью context.params.

Выполняем конвейер

Иногда вы можете захотеть поэкспериментировать с новыми функциями Python и понаблюдать за их выходом в блокноте. К счастью, Kedro позволяет запускать конвейер внутри блокнота:

Для быстрого тестирования вы также можете преобразовать функции из блокнота Jupyter в узлы Kedro.

Визуализация

Если вы запутались в структуре вашего конвейера, визуализируйте весь конвейер с помощью kedro-viz. Установите пакет:

pip install kedro-viz

Для визуализации выполните:

kedro viz

Откроется веб-браузер. Вы должны увидеть что-то вроде этого:

Визуализация отражает отношения между наборами данных и узлами. Щёлкнув по узлу, вы получите дополнительную информацию об этом узле, такую как его параметры, входные и выходные данные, путь к файлу и код.

Документирование

Kedro позволяет легко создавать документацию на основе структуры кода вашего проекта и включает все doc-string из вашего кода. Чтобы создать документацию, введите:

kedro build-docs

И документация для вашего проекта будет автоматически создана в docs/build/html! Вы увидите её, открыв docs/build/html/index.html или выполнив команду:

kedro build-docs --open

Она выглядит примерно так:

Код проекта.

Наши статьи показывают, что Data Science становится проще для новых людей. Если наука о данных интересна для вас, вы можете присмотреться к программе нашего флагманского курса, курса по аналитике данных или узнать, как начать карьеру и прокачаться в других направлениях IT:

Data Science и Machine Learning

• Профессия Data Scientist

• Профессия Data Analyst

• Курс «Математика для Data Science»

• Курс «Математика и Machine Learning для Data Science»

• Курс по Data Engineering

• Курс «Machine Learning и Deep Learning»

• Курс по Machine Learning

Python, веб-разработка

• Профессия Fullstack-разработчик на Python

• Курс «Python для веб-разработки»

• Профессия Frontend-разработчик

• Профессия Веб-разработчик

Мобильная разработка

• Профессия iOS-разработчик

• Профессия Android-разработчик

Java и C#

• Профессия Java-разработчик

• Профессия QA-инженер на JAVA

• Профессия C#-разработчик

• Профессия Разработчик игр на Unity

От основ — в глубину

• Курс «Алгоритмы и структуры данных»

• Профессия C++ разработчик

• Профессия Этичный хакер

А также:

• Курс по DevOps