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

Netflixから機械学習ワークフロー管理用のPythonライブラリ，Metaflowがリリースされました。

これを使うと，

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

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

気になる人は，tutorialを動かしつつ公式ドキュメントに目を通してみましょう。
Tutorialについては，pip install metaflowでライブラリを入れた後，

metaflow tutorials pull

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

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

ライブラリ概要

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

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

基本的な使い方

公式ドキュメントに掲載されている例を使い，単純にhello worldと表示するワークフローを動かしてみます。

ワークフローの定義

まずは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ファイルとしてまとめ，コマンドラインから

python linearflow.py run

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

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のようなものが振られています。

また，

python linearflow.py resume

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

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

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

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オブジェクトが取得できています。

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

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

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

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

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

run.data
# <MetaflowData: my_var, name>

run.data.my_var
# 'hello world'

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

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を使うことで，パラメータ別の処理を並列させることも可能です。

詳細は公式ドキュメントを参照してください。

まとめ

Metaflowでできること

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

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

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

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

現時点ではできないこと

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

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

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

実行コード

本記事で扱った実行コードはGitHubにアップしてあります。

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