AI・機械学習チームにおけるデータパイプライン構築

Transcript

1. AI・機械学習チームにおける データパイプライン構築 エムスリー株式会社 西場正浩 @m_nishiba Data Pipeline Casual Talk #1
2. 自己紹介 • 西場正浩(m_nishiba) • エムスリー株式会社 • AI・機械学習チーム / 採用チーム リーダー • 数理ファイナンス（Ph.D.） → 金融機関（クオンツ） → エムスリー（機械学習エンジニア） • 採用チームの活動 ◦ 新規プロダクトを生み出すエンジニア／プロダクト説明会
3. 話したいこと • エムスリーにおけるMLのデータパイプラインの開発方法 • luigiを使うことのメリット(おそらくairflowでも同様） • luigiを拡張したgokartによりさらに効率的に。
4. 機械学習系の開発で困ったこと • 2018年7月にAI・機械学習チーム立ち上げ。 • 最初にリリースされたシステムは課題だらけ。 • データ（学習済みモデル、前処理済みデータも含む）の 使いまわしが難しい。 • モデルの再現性が低い。 • ログ出力が適切でない。ログが読めない。 • アルゴリズムの切り替えが煩雑になる。 • class・taskの設計が良くない。 • 開発時に同じ処理を 何度も実行している。 • テストがしづらい。設計が悪い。
5. • Luigiを使う。 ◦ data pipelineの構築が簡単にできるようになる。 ◦ ログ出力がキレイになる。 ◦ Taskのインターフェイスが決まっているので設計がシンプルに。 • Luigiを拡張したGoKartを開発中。 ◦ Taskの実装が楽に。 ◦ パラメータに応じてTaskのアウトプットを管理。 ◦ Task名とIDだけで再現できる。 ◦ slackへ通知。 問題の解決方法
6. gokart等、公開しています • luigiをラップし、自分たちのニーズを満たすようなライブラリを作る ◦ https://github.com/m3dev/gokart ◦ ☆ × 9 ◦ https://gokart.readthedocs.io/en/latest/ • 共通化できるタスク群をライブラリ化する ◦ https://github.com/m3dev/redshells ◦ ☆ × 19
7. Luigi: データパイプラインを構築する。 TaskA: - param - output: task_a.csv - run: do_something TaskB: - param - requires:TaskA(param=self.param) - output: task_b.csv - run: do something TaskBはTaskBに依存する。 TaskAのパラメータ TaskAはtask_a.csvを出力する。 何かしらの処理を行なう。 inputとしてtask_a.csvを受け取る
8. Luigi: ログが読みやすい INFO: Informed scheduler that task TaskB_test_param_f20ef5457e has status PENDING INFO: Informed scheduler that task TaskA_test_param_f20ef5457e has status PENDING INFO: Done scheduling tasks INFO: Running Worker with 1 processes INFO: [pid 20187] Worker Worker(...) running TaskA(param=test_param) INFO: [pid 20187] Worker Worker(...) done TaskA(param=test_param) INFO: Informed scheduler that task TaskA_test_param_f20ef5457e has status DONE INFO: [pid 20187] Worker Worker(...) running TaskB(param=test_param) INFO: [pid 20187] Worker Worker(...) done TaskB(param=test_param) INFO: Informed scheduler that task TaskB_test_param_f20ef5457e has status DONE INFO: Worker Worker(...) was stopped. Shutting down Keep-Alive thread INFO: ===== Luigi Execution Summary ===== Scheduled 2 tasks of which: * 2 ran successfully: - 1 TaskA(...) - 1 TaskB(...) This progress looks :) because there were no failed tasks or missing dependencies ===== Luigi Execution Summary =====
9. Luigi: Taskのインターフェイスが決まる class TaskB(gokart.TaskOnKart): param = luigi.Parameter() def requires(self): return TaskA(param=self.param) def output(self): return self.make_target('task_b.pkl') def run(self): data = self.load() self.dump(data + '!!') • 左例はgokartを使っている。luigiでも同じ。 • ３つの関数を定義するだけ ◦ requires ◦ output ◦ run • 単一責任の原則が守られやすい。 • 他のメンバーが作ったタスクも再利用しやすい。
10. Luigi: データパイプライン • Taskの組合せによりデータパイプラインを構築 • データをBigQueryから取り出し、加工し、 S3等に保存する • 各タスクのinとoutはファイルとして保存される。 Task BiqQuey API DB S3
11. gokart: Taskの実装が簡単に。 class TaskB(gokart.TaskOnKart): param = luigi.Parameter() def requires(self): return TaskA(param=self.param) def output(self): return self.make_target('task_b.pkl') def run(self): data = self.load() self.dump(data + '!!') 拡張子から自動的にフォーマットを選択 load() だけでTaskAのoutputを読み込む dump() だけでoutput()に出力
12. gokart: 出力先 class TaskB(gokart.TaskOnKart): param = luigi.Parameter() def requires(self): return TaskA(param=self.param) def output(self): return self.make_target('task_b.pkl') • 設定ファイルを読み込んで、 s3またはローカルに保存する。 • Taskのパラメータやrequiresのタスクに 応じてuniqueなファイル名に変換 • ex. task_b_3d5b7a50a230d4.pkl
13. gokart: モデルの保存 def output(self): return self.make_model_target( 'word2vec.zip', save_function=gensim.models.Word2Vec.save, load_function=gensim.models.Word2Vec.load) def run(self): texts = self.load() # type: List[List[str]] shuffle(texts) model = gensim.models.Word2Vec( sentences=texts, **self.word2vec_kwargs) self.dump(model) • 複数のファイルを出力するモデルは zip にまとめて保存。 • saveやloadの関数を指定 保存はdumpを呼ぶだけ。
14. • TaskA(赤) ◦ パラメータが変更された ◦ 出力ファイルが削除された ◦ パラメータは同じだが更新された • 黄色のタスク群は再実行される。 gokart: 再実行の設定 TaskA BiqQuey API DB S3
15. gokart: Slackへの通知 • タスク同士の依存関係や実行パラメータ等が取得できる • ちゃんと仕込めばbigquery等の依存テーブル一覧を取得することも可能 ===== Event List ==== ---- Success Tasks ---- TaskB:[9a4f24468013c7daeeac64] ==== Tree Info ==== └─-(COMPLETE) TaskB[9a4f24468013c7daeeac64](parameter={'param': 'Hello'}, output=['./resources/output_of_task_b_9a4f24468013c7daeeac64.pkl'], time=0.0013031s, task_log={}) └─-(COMPLETE) TaskA[060b9dac70db9e17da7d0](parameter={'param': 'called by TaskB'}, output=['./resources/output_of_task_a_060b9dac70db9e17da7d0.pkl'], time=0.00066s, task_log={}) • タスク名とIDが分かれば手元で簡単に再現ができる • ジョブが終わったらslackに上記の情報を通知
16. gokart: pipelineの構築 class TaskA(gokart.TaskOnKart): param = luigi.Parameter() class TaskB(gokart.TaskOnKart): input_task = gokart.TaskInstanceParameter() class TaskC(gokart.TaskOnKart): def requires(self): return TaskB(input_task=TaskA(param='test')) requiresの中で複雑なパイプラインを構築で きる。 ex. word2vecからfasttextへの変更できる Taskのパラメータとして、 Taskのインスタンスを指定できる。
17. まとめ • 機械学習におけるデータパイプラインを luigiで管理している • luigiを自分たちのニーズに合わせて改造している (gokart) • タスクや結果の使い回しができるようになり、生産性が上がった（ redshells）