ちっちゃい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
  }

まとめと所感

最後に