ちっちゃいMLOpsやってみた（手書き文字認識アプリ＋kedro）

今回の実装の焦点

アプリ制作

• フロントエンド：JavaScript
• バックエンド：Flask
• 機械学習ライブラリ：Keras

kedroの導入

$ pip install kedro
$ kedro --version
kedro, version 0.17.5
$ kedro new

.
├── ...
├── conf
│   └── base
│       ├── ...
│       ├── catalog.yml           # データの登録
│       └── parameters.yml        # パラメータの登録
├── data
│   ├── ...
│   ├── 03_primary
│   │   ├── app.pkl               # アプリからのデータ
│   │   └── mnist.pkl             # MNISTのデータ
│   ├── ...
│   ├── 06_models
│   │   ├── model                 # アプリ用モデル
│   │   ├── pre_model             # 訓練後のモデル
│   │   └── tune_model            # チューニング後のモデル
│   └── ...
└── src
    ├── ...
    └── tegaki
        ├── ...
        ├── pipeline_registry.py  # 全体のパイプライン定義
        └── pipelines
            ├── data_engineering
            │   ├── __init__.py
            │   ├── nodes.py      # 前処理を記述
            │   └── pipeline.py   # 前処理のパイプライン定義
            ├── data_science
            │   ├── __init__.py
            │   ├── nodes.py      # 訓練・テストの処理を記述
            │   └── pipeline.py   # 訓練・テストのパイプライン定義
            └── fine_tuning
                ├── __init__.py
                ├── nodes.py      # チューニングの処理を記述
                └── pipeline.py   # チューニングのパイプライン定義

conf/base ディレクトリと data ディレクトリ

(catalog.yamlの内容)
mnist_data:
  type: pickle.PickleDataSet
  filepath: data/03_primary/mnist.pkl
  backend: pickle

app_data:
  type: pickle.PickleDataSet
  filepath: data/03_primary/app.pkl
  backend: pickle

pre_model:
  type: tensorflow.TensorFlowModelDataset
  filepath: data/06_models/pre_model

tune_model:
  type: tensorflow.TensorFlowModelDataset
  filepath: data/06_models/tune_model
#
# アプリで参照するモデル 今回は手動でコピーしています
# model:
#   type: tensorflow.TensorFlowModelDataset
#   filepath: data/06_models/model

(parameters.ymlの内容)
num_classes: 10
img_rows: 28
img_cols: 28
batch_size: 128
...

src ディレクトリ

(data_science内のnode.pyの一部)
def train(x_train, y_train, x_valid, y_valid, parameters):

  model = Sequential()
  model.add(Conv2D(32, kernel_size=(3, 3),activation='relu',\
    input_shape=(parameters["img_rows"],parameters["img_cols"],1)))
  model.add(Conv2D(64, (3, 3), activation='relu'))
  ...(中略)...
  model.add(Dense(parameters["num_classes"], activation='softmax'))

  model.compile(loss='categorical_crossentropy',\
    optimizer=RMSprop(),metrics=['accuracy'])
  model.fit(x_train, y_train, batch_size=parameters["batch_size"], \
    epochs=parameters["epochs"], verbose=1, validation_data=(x_valid, y_valid))

  return model

def report(model, x, y):
  score = model.evaluate(x, y, verbose=0)
  print("****************************")
  print('Test loss:', score[0])
  print('Test accuracy:', score[1])
  print("****************************")

(fine_tuning内のnode.pyの一部)
def tuning(model, x_app, y_app, parameters):
  for i in range(len(model.layers)):
    if i != parameters["tuning_layer"]:
      model.layers[i].trainable == False
  model.compile(loss='categorical_crossentropy',\
    optimizer=RMSprop(),metrics=['accuracy'])
  model.fit(x_app, y_app, batch_size=parameters["tuning_batch_size"], \
    epochs=parameters["tuning_epochs"], verbose=1)
  return model

(fine_tuning内のpipeline.py)
from kedro.pipeline import Pipeline, node

from .nodes import tuning
from ..data_science.nodes import report

def create_pipeline(**kwargs):
  return Pipeline(
    [
      node(
        func=tuning,          # node.pyに記述した関数名
        inputs=[              # パイプラインの前の関数からのインプット
            "pre_model",      # catalog.ymlに書いた"pre_model"
            "app_x",          # 前のパイプラインから得た画像データのArray
            "app_y",          # 前のパイプラインから得た正解ラベル
            "parameters"      # parameters.ymlに記述したパラメータ
        ],
        outputs="tune_model", # チューニングしたモデルを次に渡す
        name="tuning",
      ),
      node(
        func=report,          # node.pyに記述した関数名
        inputs=[
            "tune_model",     # チューニングしたモデルを受け取る
            "mnist_test_x",   # 前のパイプラインから得たMNISTのテストデータセット
            "mnist_test_y",   # 前のパイプラインから得たMNISTのテストデータセット
        ],
        outputs=None,
        name="report"
      )
    ]
  )

from kedro.pipeline import Pipeline
from tegaki.pipelines import data_engineering as de
from tegaki.pipelines import data_science as ds
from tegaki.pipelines import fine_tuning as ft

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

  ...(中略)...

  tegaki_de_pipeline = de.create_pipeline()
  tegaki_ds_pipeline = ds.create_pipeline()
  tegaki_ft_pipeline = ft.create_pipeline()

  return {
    "de": tegaki_de_pipeline,
    "ds": tegaki_ds_pipeline,
    "ft": tegaki_ft_pipeline,
    "__default__": tegaki_de_pipeline + tegaki_ds_pipeline,
    "tuning": tegaki_de_pipeline + tegaki_ft_pipeline
  }

それぞれの pipeline を呼び出して、上述のように書くことで実行できるようになります。今回の場合、kedro runを実行するとデータを前処理して訓練してテストして結果を出力、kedro run --pipeline tuningを実行するとデータを前処理してモデルを読み込んでファインチューニングしてテストして結果を出すという一連の流れをやってくれます。このようにしてパイプラインを構築していきます。

まとめと所感

最後に