Netflix開発のMLワークフローツール，Metaflowを試してみた【概要編】

Netflixから機械学習ワークフロー管理用のPythonライブラリ，[Metaflow](https://metaflow.org)がリリースされました。

これを使うと，

- データ処理・モデル構築プロセスを統一フォーマットで記述でき，全体のフローを追いやすい
- モデル・前処理工程のバージョン管理ができる
- AWS環境上での分散処理が可能

といったメリットがあります。

気になる人は，[tutorial](https://docs.metaflow.org/getting-started/tutorials)を動かしつつ[公式ドキュメント](https://docs.metaflow.org)に目を通してみましょう。
Tutorialについては，`pip install metaflow`でライブラリを入れた後，

```Shell
metaflow tutorials pull
```
とするだけで一式揃いますので，気軽に試すことができます。

本記事では，ざっくりとした機能概要と使い方をまとめていきたいと思います。

# ライブラリ概要

Metaflowでは，データ処理や機械学習モデル構築・予測のワークフローをPythonのクラスとして定義し，コマンドラインから実行します。

その際実行の都度IDが振られ，実行結果や中間生成オブジェクトがまとめて保存されます。
これにより，実行結果を後から個別に参照可能なほか，処理の一部分や構築したモデルを別のフローで再利用，プロセスの再実行といったことも容易に行えます。

# 基本的な使い方

[公式ドキュメント](https://docs.metaflow.org)に掲載されている[例](https://docs.metaflow.org/metaflow/basics)を使い，単純に`hello world`と表示するワークフローを動かしてみます。

## ワークフローの定義

まずはPythonで下記のような処理フローを定義します。
一つのフローは一つのクラスで書き下し，各ステップはメソッドで定義します。

```Python:linearflow.py
from metaflow import FlowSpec, step

class LinearFlow(FlowSpec):

    @step
    def start(self):
        self.my_var = 'hello world'
        self.next(self.a)

    @step
    def a(self):
        print('the data artifact is: %s' % self.my_var)
        self.next(self.end)

    @step
    def end(self):
        print('the data artifact is still: %s' % self.my_var)

if __name__ == '__main__':
    LinearFlow()
```

基本的な書き方は，

- クラスに`FlowSpec`を継承させる
- 各メソッドにデコレータ`@step`を付記する
    - メソッド名は，`start`から始まり`end`で終わる
    - 中間ステップのメソッド名は自由に決められる。
- 各メソッドの終わりで`next()`により後続ステップを指定する

のようになっています。

上記の例では，`start` -> ステップ`a` -> `end`の順に実行し，`start`で作成した変数`my_var='hello world'`を`a`と`end`で表示します。

## コマンドライン実行

上記を`linearflow.py`ファイルとしてまとめ，コマンドラインから

```Shell
python linearflow.py run
```

のように起動します。
すると，各ステップが順に実行され，以下のように表示されます：

```Commandline
Metaflow 2.0.0 executing LinearFlow for user:tatamiya
Validating your flow...
    The graph looks good!
Running pylint...
    Pylint not found, so extra checks are disabled.
2019-12-08 13:14:36.636 Workflow starting (run-id 1575778476628625):
2019-12-08 13:14:36.639 [1575778476628625/start/1 (pid 33549)] Task is starting.
2019-12-08 13:14:36.896 [1575778476628625/start/1 (pid 33549)] Task finished successfully.
2019-12-08 13:14:36.900 [1575778476628625/a/2 (pid 33554)] Task is starting.
2019-12-08 13:14:37.129 [1575778476628625/a/2 (pid 33554)] the data artifact is: hello world
2019-12-08 13:14:37.162 [1575778476628625/a/2 (pid 33554)] Task finished successfully.
2019-12-08 13:14:37.166 [1575778476628625/end/3 (pid 33559)] Task is starting.
2019-12-08 13:14:37.384 [1575778476628625/end/3 (pid 33559)] the data artifact is still: hello world
2019-12-08 13:14:37.416 [1575778476628625/end/3 (pid 33559)] Task finished successfully.
2019-12-08 13:14:37.416 Done!
```

この際，上記では`1575778476628625`という実行IDのようなものが振られています。

また，

```Shell
python linearflow.py resume
```

とすると，直前に実行した内容を再実行できます。
この際，実行IDは前回とは異なるものが得られます。


## 実行時のオブジェクトを再取得

過去の実行内容は`Flow`モジュールを使って参照できます。

```Python
from metaflow import Flow

for run in Flow('LinearFlow').runs():
    print(run)
# Run('LinearFlow/1575778694363290')
# Run('LinearFlow/1575778644025078')
# Run('LinearFlow/1575778633533099')
# Run('LinearFlow/1575778476628625')
```

上記では，`resume`による再実行含めて4回起動したので，4つの`Run`オブジェクトが取得できています。

このうち，一番最初に実行した実行ID`1575778476628625`の結果を取得するには，
`Flow`オブジェクトを一旦リスト化して末尾のものを取得する，

```Python
run = list(Flow('LinearFlow'))[-1]
```

もしくは，`Run`モジュールから直接指定します：

```Python
from metaflow import Run
run = Run('LinearFlow/1575778476628625')
```

この時の`Run`オブジェクトの`data`プロパティを見ると，クラス内で定義されていた変数`my_var`を再取得することができます。

```Python
run.data
# <MetaflowData: my_var, name>

run.data.my_var
# 'hello world'
```

なお，`Run`オブジェクトからは，実行時刻，実行環境などのメタ情報も含まれています。

```Python
run.created_at
# '2019-12-08T04:14:36.fZ'

run.tags
# frozenset({'date:2019-12-08',
#           'metaflow_version:2.0.0',
#           'python_version:3.7.4',
#           'runtime:dev',
#           'user:tatamiya'})
```

ちなみにこうした中間オブジェクトやメタ情報は，実行ディレクトリ内の`.metaflow/`以下に全て保存されています。

## ステップ分岐について

今回作成したDAGでは分岐のない直列プロセスでしたが，途中で`self.next()`に複数ステップ指定することで，依存関係のない複数ステップをパラで走らせて後から結果を統合することもできます。

または，`self.next()`のオプション引数`foreach`を使うことで，パラメータ別の処理を並列させることも可能です。

詳細は[公式ドキュメント](https://docs.metaflow.org/metaflow/basics)を参照してください。

# まとめ

## Metaflowでできること

ざっと調べてみた限りでは，Metaflowでは以下のことができました。

- ワークフロー定義
- 処理の再実行
- モデル・中間生成物のバージョン管理

また，本記事では紹介できませんでしたが，下記のことも可能なようです。

- コマンドライン実行時に引数からParameterを指定
- プロセスの一部をAWS環境上で分散処理する
- 各ステップを異なる仮想環境で実行する(`conda`使用)

## 現時点ではできないこと

一方で，個人的に期待していた下記の機能は，現時点では備わっていませんでした：

- ワークフローの可視化(cf. luigi, Airflow)
- モデル精度比較用のダッシュボード表示(cf. MLflow)

これらの機能もゆくゆくは実装されるかもしれませんが，
公式ドキュメントの[Roadmap](https://docs.metaflow.org/introduction/roadmap)
を見る限りでは，R版の開発や本番システムへのデプロイ機能に重点が置かれるようです。

## 実行コード

本記事で扱った実行コードは[GitHub](https://github.com/tatamiya/metaflow_trial/tree/master/linearflow)にアップしてあります。

今回のシンプルな実行例の他に，Titanicデータを用いた予測ワークフローも一部コードを上げています。
こちらについても，改めて解説記事を作成したいと考えております。