PythonとLuigiによるデータパイプライン構築

# はじめに
この記事は [Building Data Pipelines with Python and Luigi](http://marcobonzanini.com/2015/10/24/building-data-pipelines-with-python-and-luigi/) の和訳です。
元記事がよくできていたので、自分の理解も兼ねて、拙いながらも訳してみました。
誤り等ありましたらぜひコメントでご指摘おねがいいたします。

# PythonとLuigiによるデータパイプライン構築
データサイエンティストにとって、日々の業務はエンジニアリングというよりも研究開発の色が濃いことがしばしばある。にもかかわらず、プロトタイプから製品までのプロセスには、素早く泥臭い決定が次善の策[^1]であり、けっこうな数のリエンジニアリングの努力が必要となる。これはいつも革新を遅らせ、一般的に言えば、プロジェクト全体を遅らせる。

この記事はデータパイプライン構築の経験について議論する：データの抽出、洗浄、結合、前処理などデータ駆動製品のためのデータを準備するのに必要となる一般的なすべてのステップである。特に焦点を当てるのは**データの配管**、そしてLuigiのようなワークフローマネージャがいかにあなたを邪魔せずに救世主となるのかである。最小限の努力で、プロトタイプから製品への移行がスムーズになる。

コードのサンプルは [GitHub Gist](https://gist.github.com/bonzanini/40774bf9348d35d9ea4f) で公開している。

## これまでのプロトタイプ
過去のプロトタイプでは、データパイプラインはおおむね次のような感じだった：

```bash
$ python get_some_data.py
$ python clean_some_data.py
$ python join_other_data.py
$ python do_stuff_with_data.py
```
データプロジェクトの予備実験段階では、次に挙げるようなことはごく普通のことだ：前処理が必要であり、それは素早いハック[^2]へとなりそうであり、だからエンジニアリングのベストプラクティスに煩わされ、そしてスクリプトの数は膨れ上がりデータパイプラインが牙をむく。

このアプローチは素早くそしてハックであるという利点だけを持つ。欠点は、退屈であることだ：毎回パイプラインを再実行したくなり、スクリプトの束を次々に手動で呼び出す必要がある。さらに、このプロトタイプを同僚に共有するとき、誤解の余地が大いに存在する (「なぜ`do_stuff_with_data`が動かないの？」「最初に`clean_some_data`をした？」などなど)。

明らかにハックな解は、すべてをひとつのスクリプトに押し込むことにみえる。いくらかの素早いリファクタリングの後、`do_everything.py` スクリプトは次のようになるだろう：

```py
if __name__ == '__main__':
    get_some_data()
    clean_some_data()
    join_other_data()
    do_stuff_with_data()
```
実行するのは簡単だ：

```bash
$ python do_everything.py
```
(注意：スクリプトの束を順番に呼び出すようなbashスクリプトにすべてをまとめることもできるが、欠点は変わらず同じである)

## コードのテンプレ (鋳型)
製品への準備段階となるパイプラインへと移った時、例の全てを実行するコードの側面について少し考える必要がある。特に、エラー処理が考慮されるべきだ：

```py
try:
    get_some_data()
except GetSomeDataError as e:
    # エラー処理
```
しかし全てのタスクを一緒にすると、try/exceptのクリスマスツリーと化す：

```py
try:
    get_some_data()
    try:
        clean_some_data()
        try:
            # ここでなにかをする...
        except EvenMoreErrors:
            # ...
    except CleanSomeDataError as e:
        # CleanSomeDataError を処理
except GetSomeDataError as e:
    # GetSomeDataError を処理
```
別の重要な考えるべき点はどうやってパイプラインを復帰させるかだ。たとえば、最初のいくつかのタスクは完了したが、その途中でエラーが起きたら、どうやって最初の成功したステップを再実行せずにそのパイプラインを再実行できるだろうか？

```py
# タスクがすでに成功しているかチェック
if not i_got_the_data_already():
    # していないなら、それを実行
    try:
        get_some_date()
    except GetSomeDataError as e:
        # エラー処理
```

## Luigi へ
[Luigi](https://github.com/spotify/luigi) は、バッチジョブの複雑なデータパイプライン構築を助けるためにSpotifyで開発された、ワークフロー管理のためのPythonのツールである。Luigi のインストールは：

```bash
pip install luigi
```
Luigi の便利な特徴は：

- 依存管理
- チェックポイント / 障害復帰
- CLIインテグレーション / パラメータ化
- 依存グラフ可視化

データパイプラインにどうやって Luigi を適用できるかを理解するためのキーコンセプトがふたつある：タスクとターゲットだ。タスクは仕事の集まりであり、`luigi.Task` クラスを継承、いくつかの基礎的なメソッドをオーバーライドすることで表現する。タスクの出力がターゲットであり、それはローカルファイルシステムかもしれないし、Amazon S3 かもしれないし、データベースにあるかもしれない。

依存関係は入力と出力で定義できる。例えば、もしタスクBがタスクAに依存している場合、それはタスクAの出力がタスクBの入力である、ということを意味する。

いくつかの典型的なタスクを見てみよう：

```py
# Filename: run_luigi.py
import luigi
 
class PrintNumbers(luigi.Task):
 
    def requires(self):
        return []
 
    def output(self):
        return luigi.LocalTarget("numbers_up_to_10.txt")
 
    def run(self):
        with self.output().open('w') as f:
            for i in range(1, 11):
                f.write("{}\n".format(i))
 
class SquaredNumbers(luigi.Task):
 
    def requires(self):
        return [PrintNumbers()]
 
    def output(self):
        return luigi.LocalTarget("squares.txt")
 
    def run(self):
        with self.input()[0].open() as fin, self.output().open('w') as fout:
            for line in fin:
                n = int(line.strip())
                out = n * n
                fout.write("{}:{}\n".format(n, out))
                 
if __name__ == '__main__':
    luigi.run()
```
このコードはふたつのタスクを提示している：1から10までの数を一行ずつ`number_up_to_10.txt`というファイルに書き出す`PrintNumbers`と、そのファイルを読んで平方数とのペアを一行ずつ`squares.txt`というファイルに書き出す`SquaredNumbers`だ。

このタスクを実行するには：

```bash
$ python run_luigi.py SquaredNumbers --local-scheduler
```
Luigi はタスク間の依存関係チェックを考慮し、`SquaredNumbers`の入力がないことを発見するので、最初に`PrintNumbers`タスクを走らせてから、`SquaredNumbers`を実行に移す。

Luigi に渡した最初の引数は実行したいパイプラインの最後のタスクの名前だ。二番目の引数は、単に Luigi にローカルスケジューラを使うように伝えているだけだ (詳細は後述)。

`luigi` コマンドを用いることもできる：

```bash
$ luigi -m run_luigi.py SquaredNumbers --local-scheduler
```

## タスクの骨組み
Luigi のタスクを作るためには、単に`luigi.Task`を親としてクラスを作り、いくつかのメソッドをオーバーライドすればよい。特に：

- `requires()`は依存しているタスクのリスト
- `output()`はタスクのターゲット (例えば LocalTarget, S3Targetなど)
- `run()`は実行のロジック

である。Luigi は`requires()`と`output()`の返り値をチェックし、それに応じて依存関係グラフを構築する。

## パラメータを渡す
ハードコーディングされたファイル名や設定値は一般的に言ってアンチパターンだ。一度タスクの構造とダイナミクスを理解したら、同じスクリプトを引数を変えて動的に呼び出すことが出来るように設定をパラメータ化すべきだ。

`luigi.Parameter()`クラスがそれだ。それぞれの Luigi のタスクはいくつかのパラメータを持ちうる。たとえば、前の例で、数を変えられるようにするとしよう。`range()`関数のパラメータとして使用しているのは整数なので、デフォルトのパラメータクラスではなく`luigi.IntParameter`を使うことができる。変更したタスクは次のようになる：

```py
class PrintNumbers(luigi.Task):
    n = luigi.IntParameter()
 
    def requires(self):
        return []
 
    def output(self):
        return luigi.LocalTarget("numbers_up_to_{}.txt".format(self.n))
 
    def run(self):
        with self.output().open('w') as f:
            for i in range(1, self.n+1):
                f.write("{}\n".format(i))
 
class SquaredNumbers(luigi.Task):
    n = luigi.IntParameter()
 
    def requires(self):
        return [PrintNumbers(n=self.n)]
 
    def output(self):
        return luigi.LocalTarget("squares_up_to_{}.txt".format(self.n))
 
    def run(self):
        with self.input()[0].open() as fin, self.output().open('w') as fout:
            for line in fin:
                n = int(line.strip())
                out = n * n
                fout.write("{}:{}\n".format(n, out))
```
`SquaredNumbers`タスクを20まで上げて呼び出すには：

```bash
$ python run_luigi.py SquaredNumbers --local-scheduler --n 20
```
パラメータにはデフォルト値を持たせることもできる。例えば：

```py
n = luigi.IntParameter(default=10)
```
この場合、`--n`引数を指定しなければ、10が使われる。

[サンプルはGitHub Gistへ](https://gist.github.com/bonzanini/40774bf9348d35d9ea4f)

## ローカル vs グローバルスケジューラ
前に、Luigi のタスクをローカルスケジューラで実行するときに`--local-scheduler`オプションを使った。これは開発には便利だが、製品環境の場合、centralised scheduler を使うべきだ ([scheduler](http://luigi.readthedocs.org/en/latest/central_scheduler.html) のドキュメントを見よ)。

これにはいくつかの利点がある：

- 同じタスクのふたつのインスタンスを同時に実行することを回避
- ナイスなWebベースでの可視化

Luigi スケジューラのデーモンをフォアグラウンドで走らせるには：

```bash
$ luigid
```
バックグラウンドの場合：

```bash
$ luigid --background
```
これはデフォルトで8082番ポートを使うので、ブラウザで http://localhost:8082/ へアクセスすれば可視化を見ることができる。

グローバル Luigi スケジューラが走っている時は、ローカルスケジューラのためのオプションなしで再実行できる：

```bash
$ python run_luigi.py SquaredNumbers --n [BIG_NUMBER]
```
サンプルコードはミリ秒単位で終了するが、もしブラウザに切り替えてタスクがまだ実行中のときの依存関係グラフを見たいなら、おそらく大きな数たとえば10,000,000とかそれ以上の数を`--n`オプションに与えれば良いだろう。

依存関係グラフのスクリーンショットは：
<img width="736" alt="dependency-graph-screenshot.png" src="https://qiita-image-store.s3.amazonaws.com/0/62450/b658fe99-b830-eeef-3a22-f23a3891318c.png">

## まとめ
Pythonで書かれたワークフローマネージャ、Luigi を使ったデータパイプラインの定義について議論した。Luigi はタスクとターゲットによるパイプラインのナイスな抽象化を提供し、あなたのために依存関係も考慮してくれる。

コード再利用、そしてプロトタイプから製品へ移行するマインドセットの観点から、私はビジネスロジックのタスクを個別の[Pythonパッケージ](http://marcobonzanini.com/2015/07/01/how-to-develop-and-distribute-python-packages/)として定義することが便利だと思う (つまり、`setup.py`ファイルとともに)。このやり方だと、Luigi のスクリプトからは単に `import your_package` と宣言してそこから呼び出せば良い。

ひとつのタスクが複数のファイルを出力として生成することは可能だが、もしそうなった場合、おそらくだが、そのタスクが小さな単位 (つまり複数のタスク) に分割できないかを考えるべきだ。それらの出力は論理的に同じか？依存関係はないか？もしタスクを分割できない場合、ただ`output()`をタスクそれ自体の名前やらタイムスタンプやらを組み合わせたログファイルとするのが単純かつ便利だと思う。そのログファイルの名前は`TaskName_timestamp_param1value_param2value_etc`のようになるだろう。

Luigi のようなワークフローマネージャは、データパイプラインを開発するときに、依存関係を処理し、パラメータやエラー処理のためのコードの鋳型の量を減らし、障害復帰を管理し、明快なパターンに従わせてくれるので、一般的に言って有用だ。

限界も考慮することが重要である：

- Luigi はバッチジョブのために開発されているので、リアルタイムに近い処理にはたぶん役に立たない
- 実行をトリガーしてはくれない。あなたはデータパイプラインを実行する必要がある (例えば cronjob を通して)

---

[^1]: 原文："some of the early quick-and-dirty decisions turn out to be sub-optimal"
[^2]: 原文："quick hack"