thunderbolt⚡ on gokartでパイプラインの出力結果をサクサク確認する

# これなに?

F81アドベントカレンダー9日目の長谷川です。明日の担当は本日が誕生日🎉の [pastelInc](https://qiita.com/pastelInc)さんです。おめでとうございます。  
この記事では、パイプラインフレームワークの[Luigi](https://github.com/spotify/luigi)の出力結果を、Jupyterなどで簡単に確認するためのライブラリ、[thunderbolt](https://github.com/m3dev/thunderbolt)の紹介していきます。   
なお、この記事ではLuigiの基礎的な内容については、一切解説しません。

## パイプライン処理の課題

パイプラインフレームワークを使ってデータ処理を開発、運用しているときに、よく**中間生成物を簡単に確認したい！** と思うこと多々あります。    
有向非巡回グラフの特性上、依存している処理の出力結果が期待している物とズレていると、そのズレが連鎖的に伝搬し、ゴミを入れてもゴミしか出ない、汚染されたパイプラインになりかねないからです。  
Taskを追加しまくって、どんどん開発が進むのは気分が良いですが、出力物の確認をせずにガンガン処理を増やしていくと、パイプラインが複雑化し、どこでどのような処理が行われているのかが徐々に不透明になり、期待した結果とのズレが生じた際に、そのズレがどこで生じているのかを特定するのが、より困難になっていきます。

# thunderbolt⚡とは?

そこでthunderbolt(とgokart)の出番です。  
thunderboltは、Luigiをより使いやすくしたラッパーライブラリである[gokart](https://github.com/m3dev/gokart)で出力されるlogをパースし、マネージドできるライブラリです。名前の由来はマリオカートのアイテム、[サンダー](https://www.nintendo.co.jp/3ds/amkj/item/thunder.html)⚡からですね。  
名の通り、あくまでgokartをサポートするツールなので、**gokartでluigiを構築していることが前提となっています。**  gokartの使い方は、[M3さんのブログ](https://www.m3tech.blog/entry/2019/09/30/120229)か、[前に書いた記事](https://qiita.com/Hase8388/items/8cf0e5c77f00b555748f)をお読みください。

## gokartが出力するlog

gokartは以下のように`.pkl`形式でlogを出力します。

```
 resources
 ├── data_3eba828ec57403111694b0fb1e3f62e0.pkl
 └── log
     ├── module_versions
     │   └── TaskB_3eba828ec57403111694b0fb1e3f62e0.txt
     ├── processing_time
     │   └── TaskB_3eba828ec57403111694b0fb1e3f62e0.pkl
     ├── task_log
     │   └── TaskB_3eba828ec57403111694b0fb1e3f62e0.pkl
     └── task_params
         └── TaskB_3eba828ec57403111694b0fb1e3f62e0.pkl
```



logには、

- 利用したモジュールのバージョン
- 処理にかかった時間
- loggerで出力されたlog
- 適用されたパラメータ

が保存されています。

gokartの場合、デフォルトだと指定したファイル名にhash値がついて保存されてます。hash値はTaskのパラメータによって決まるため頻繁に変わり、しかもlogは`pickle`で保存されるので、出力物をloadしたり、ログを確認したりが、いささか面倒ですが、このあたりをサポートしてくれるのがthunderboltです。

# 使い方

[公式のexample](https://github.com/m3dev/thunderbolt/blob/master/examples/example.ipynb)に載ってます、おわり。  
だと、ちょっと寂しいので、基本的な使い方をここでも紹介します

なおライブラリのバージョンは

- luigi==2.8.10
- gokart==0.3.7
- thunderbolt==0.0.1

です

## 初期化


```python
from thunderbolt import Thunderbolt
tb=Thunderbolt(workspace_directory='[goartの保存先のディレクトリ]')
```

で、マネージャのインスタンスを作成します。

なお、ディレクトリのパラメータが指定されていない場合、環境変数`TASK_WORKSPACE_DIRECTORY`で指定されたpathが参照されます。

## 出力物の確認

以下のように`get_task_df`で読み込むと、出力物の情報が`DataFrame`として確認できます。


```python
tb.get_task_df()
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>task_id</th>
      <th>task_name</th>
      <th>last_modified</th>
      <th>task_params</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>0</th>
      <td>0</td>
      <td>LoadTargetData</td>
      <td>2019-11-20 11:04:22.727454</td>
      <td>{'target_id': '0'}</td>
    </tr>
    <tr>
      <th>1</th>
      <td>1</td>
      <td>LoadData</td>
      <td>2019-11-20 11:04:22.794075</td>
      <td>{}</td>
    </tr>
    <tr>
      <th>2</th>
      <td>2</td>
      <td>LoadTargetData</td>
      <td>2019-11-20 11:04:22.774771</td>
      <td>{'target_id': '2'}</td>
    </tr>
    <tr>
      <th>3</th>
      <td>3</td>
      <td>TargetId</td>
      <td>2019-11-20 11:04:22.669280</td>
      <td>{}</td>
    </tr>
  </tbody>
</table>
</div>



## 出力物をload

以下のように`load(task_id=[確認したい出力物のtask_id])`で出力物の中身を確認できます。 


```python
tb.load(task_id=1)[0].head()
```




<div>
<style scoped>
    .dataframe tbody tr th:only-of-type {
        vertical-align: middle;
    }

    .dataframe tbody tr th {
        vertical-align: top;
    }

    .dataframe thead th {
        text-align: right;
    }
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>0</th>
      <th>1</th>
      <th>2</th>
      <th>3</th>
      <th>target</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>100</th>
      <td>6.3</td>
      <td>3.3</td>
      <td>6.0</td>
      <td>2.5</td>
      <td>2</td>
    </tr>
    <tr>
      <th>101</th>
      <td>5.8</td>
      <td>2.7</td>
      <td>5.1</td>
      <td>1.9</td>
      <td>2</td>
    </tr>
    <tr>
      <th>102</th>
      <td>7.1</td>
      <td>3.0</td>
      <td>5.9</td>
      <td>2.1</td>
      <td>2</td>
    </tr>
    <tr>
      <th>103</th>
      <td>6.3</td>
      <td>2.9</td>
      <td>5.6</td>
      <td>1.8</td>
      <td>2</td>
    </tr>
    <tr>
      <th>104</th>
      <td>6.5</td>
      <td>3.0</td>
      <td>5.8</td>
      <td>2.2</td>
      <td>2</td>
    </tr>
  </tbody>
</table>
</div>



## その他
今回は試してないですが、S3とGCSもサポートしているので、ディレクトリのpathを`s3://xxx`あるいは`gs://xxx`とバケットを指定すると、クラウド上に出力物が保存されていても問題なく実行できるはずです。

# 終わりに
パイプラインジャングルを彷徨った際に、⚡の光が一縷の望みにならんことを！