Kedroとは

はじめに

Kedroとは、再現性・保守性・モジュール性に優れたデータサイエンスコードを作成するためのオープンソースのPythonフレームワークである。本稿では、Kedroの入門を行う。

Kedroを用いた開発フロー

公式ドキュメントによれば、Kedroを用いた開発フローは以下の通りである。

プロジェクトテンプレートを作る
データフォルダを設定する
パイプラインを作成する
プロジェクトをパッケージ化する

本稿では主に1,2,3について具体的に進める。

仮想環境の設定（Optional）

kedroの依存ライブラリをインストールする都合上、conda等であらかじめkedro用の仮想環境を作るとよい。

anacondaであれば、

$ conda create --name kedro-env python=3.7 -y
$ conda activate kedro-env

で仮想環境"kedro-env"を作成し、移動することができる。

1.プロジェクトテンプレートを作る

まずkedroをpipでインストールする

$ pip install kedro

次に、kedroプロジェクトを作成する。コマンドライン上で

$ kedro new

と入力すると新規プロジェクトが作成される。

しかし本稿ではIrisデータを使ってチュートリアルを行いたいので、starterコマンドを追加することにする。これにより、あらかじめ用意されているIris用テンプレートを使用することが出来る。

$ kedro new --starter=pandas-iris

次にProject NameとRepository Nameなどの入力が求められるが、任意の文字列を入力する（ただし、Project Nameには"-"などの文字が使えない）。ここでは、

# Project Name:
$ get_started

# Repository Name:
$ get-started

Python Package Nameに関しては、特に何も入力しないことにする。この時、以下のような構成のフォルダが作成される。

get-started
│  .gitignore
│  pyproject.toml
│  README.md
│  setup.cfg
│
├─.ipython
│  └─profile_default
│          ipython_config.py
│
├─conf
│  │  README.md
│  │
│  ├─base
│  │      catalog.yml
│  │      logging.yml
│  │      parameters.yml
│  │
│  └─local
│          .gitkeep
│          credentials.yml
│
├─data
│  ├─01_raw
│  │      .gitkeep
│  │      iris.csv
│  │
│  ├─02_intermediate
│  │      .gitkeep
│  │
│  ├─03_primary
│  │      .gitkeep
│  │
│  ├─04_feature
│  │      .gitkeep
│  │
│  ├─05_model_input
│  │      .gitkeep
│  │
│  ├─06_models
│  │      .gitkeep
│  │
│  ├─07_model_output
│  │      .gitkeep
│  │
│  └─08_reporting
│          .gitkeep
│
├─docs
│  └─source
│          conf.py
│          index.rst
│
├─logs
│  │  .gitkeep
│  │
│  └─journals
│          .gitkeep
│
├─notebooks
│      .gitkeep
│
└─src
    │  requirements.txt
    │  setup.py
    │
    ├─get_started
    │  │  hooks.py
    │  │  pipeline_registry.py
    │  │  settings.py
    │  │  __init__.py
    │  │  __main__.py
    │  │
    │  └─pipelines
    │      │  __init__.py
    │      │
    │      ├─data_engineering
    │      │      nodes.py
    │      │      pipeline.py
    │      │      README.md
    │      │      __init__.py
    │      │
    │      └─data_science
    │              nodes.py
    │              pipeline.py
    │              README.md
    │              __init__.py
    │
    └─tests
        │  test_run.py
        │  __init__.py
        │
        └─pipelines
                __init__.py

data/01_rawにiris.csvが置いてあるのが分かる。次に、依存ライブラリをインストールする。

$ cd get-started

でディレクトリを移動し、

$ kedro install

でsrc/requirement.txtに記述されているファイルがインストールされる。次に、実験的にパイプラインを実行してみる。

$ kedro run

でsrc/get-started以下の.pyファイルが実行される。

2.データフォルダを設定する

使用するcsvファイルをconf/base/catalog.ymlに設定する。入力する情報は

ファイルパス（filepath）
データセットに必要なパラメータ
データ形式（type）
バージョン

の4つであるが、今回はその内ファイルパスとデータ形式のみ入力することにする。

今回の場合、iris.csvファイルがすでにexample_iris_dataとして以下の様に設定されている。

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

新たにcsvファイルを登録したい場合は、同様の形式で入力する。ここでは、test.csvファイルを作成し、01_raw/以下に置いたうえで、catalog.ymlに以下のように追記する。

test:
  type: pandas.CSVDataSet
  filepath: data/01_raw/test.csv

データがKedroに認識されているか確認するためには、Kedro上でipythonを起動してみるとよい。コマンドライン上で

$ kedro ipython

を入力し、

companies = catalog.load("test")
companies.head()

とすれば確認出来る。

3.パイプラインを作成する

コマンドライン上で

$ kedro run

と入力するとsrc/get-started/*.pyが実行される。

今回は特にsrc/get-started/pipeline_registry.pyが重要であり、ここにパイプラインの流れが記載されている。

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

from kedro.pipeline import Pipeline

from get_started.pipelines import data_engineering as de
from get_started.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,
    }

ここではsrc/get_started/pipelines/data_engineeringとsrc/get_started/pipelines/data_scienceを呼び出し、それぞれインスタンス化している。

さらに、各サブフォルダ（data_engineering, data_science）の中のpipeline.pyではnode.pyを呼び出しており、実際の処理（前処理）はここに書かれている。

尚、出力されたデータはプログラムの実行が終わるとメモリから消えてしまうが、当然ながらフォルダに保存することも出来る。

今回のirisデータでは、 src\get_started\pipelines\data_engineering\pipeline.pyで example_train_xなどの前処理済みデータを出力しているが、試しにこれをdata/02_intermediate/に保存してみる。

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