Hello Worldタスクを作ってみた | Luigi Advent Calendar 2016 #01

はじめに

好物はインフラとフロントエンドのかじわらゆたかです。
今までETL処理はほとんどTalendで実装してきたのですが、
無償版であるOpenStudioでの実装だとどうしても変更の差分を追いかけづらいと言ったことがありました。
実装者が１人で案件をまわしているうちは良いのですが、
複数人になってくるとGitなどのソース管理の仕組みで変更箇所を追いかけたいものです。
そういった悩みを解決できるのではと思い、今回Luigiに手を出してみました。
ちょうどクリスマスシーズンですので、勉強してきた内容を１人アドベントカレンダーとしてアウトプットしていきたいと思います。

Luigiって何？

Githubのリポジトリには以下のように記載されております。 Luigi is a Python module that helps you build complex pipelines of batch jobs. It handles dependency resolution, workflow management, visualization etc. It also comes with Hadoop support built in. 和訳してみると以下のような感じでしょうか。
Luigiはバッチジョブの複雑なパイプラインを構築することのできるPythonのモジュールです。 Luigiは依存性の解決・ワークフロー管理・可視化を機能として持っています。また、Hadoopのサポートも組み込まれているとのことです。

Pythonで書けるとのことなので、Gitなどで変更箇所を把握したりすることも容易にできそうです。

Hello World

まずはHello Worldを表示させるところからやっていきたいと思います。
以前の記事でPythonのデバッグ環境の構築をやったので、Pythonの環境はその時の環境を使っていきます。
Visual Studio CodeでPythonの開発環境構築を構築してみた。 ｜ Developers.IO

まずはluigiを入れる必要があるので、pipでluigiを導入します。

$ pip install luigi

次にluigiのtaskは以下のようになります。

HelloWorld.py

import luigi


class HelloWorld(luigi.Task)

    def run(self):
        print("Hello world.")

最後に実行する際のコマンドですが、今回はCentral Schedulerに登録はしないので、Local Schedulerで動かす用にオプションを付与し、
作成したファイル名・クラス名をluigiに渡して上げれば良いとそう思っていました。

$ luigi --module HelloWorld HelloWorld --local-scheduler

上記で動かそうとしたところ、moduleが見つからないと行ったエラーとなってしまいます。

$ luigi --module HelloWorld HelloWorld --local-scheduler
Traceback (most recent call last):
  File "/Users/kajiwarayutaka/.pyenv/versions/luigiStudy/bin/luigi", line 11, in <module>
    sys.exit(luigi_run())
  File "/Users/kajiwarayutaka/.pyenv/versions/2.7.11/envs/luigiStudy/lib/python2.7/site-packages/luigi/cmdline.py", line 11, in luigi_run
    run_with_retcodes(argv)
  File "/Users/kajiwarayutaka/.pyenv/versions/2.7.11/envs/luigiStudy/lib/python2.7/site-packages/luigi/retcodes.py", line 69, in run_with_retcodes
    with luigi.cmdline_parser.CmdlineParser.global_instance(argv):
  File "/Users/kajiwarayutaka/.pyenv/versions/2.7.11/lib/python2.7/contextlib.py", line 17, in __enter__
    return self.gen.next()
  File "/Users/kajiwarayutaka/.pyenv/versions/2.7.11/envs/luigiStudy/lib/python2.7/site-packages/luigi/cmdline_parser.py", line 52, in global_instance
    new_value = CmdlineParser(cmdline_args)
  File "/Users/kajiwarayutaka/.pyenv/versions/2.7.11/envs/luigiStudy/lib/python2.7/site-packages/luigi/cmdline_parser.py", line 64, in __init__
    self._attempt_load_module(known_args)
  File "/Users/kajiwarayutaka/.pyenv/versions/2.7.11/envs/luigiStudy/lib/python2.7/site-packages/luigi/cmdline_parser.py", line 144, in _attempt_load_module
    __import__(module)
ImportError: No module named HelloWorld

これ実はLuigiあるあるのようで、カレントディレクトリのパスをPYTHONPATHに含ませて上げる必要があるとのことです。

Example – Top Artists — Luigi 2.3.3 documentation
そのため正しくは以下のようになります。

$ PYTHONPATH='.' luigi --module HelloWorld HelloWorld --local-scheduler
DEBUG: Checking if HelloWorld() is complete
/Users/kajiwarayutaka/.pyenv/versions/2.7.11/envs/luigiStudy/lib/python2.7/site-packages/luigi/worker.py:295: UserWarning: Task HelloWorld() without outputs has no custom complete() method
  is_complete = task.complete()
INFO: Informed scheduler that task   HelloWorld__99914b932b   has status   PENDING
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 37061] Worker Worker(salt=134305396, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=37061) running   HelloWorld()
Hello, world!
INFO: [pid 37061] Worker Worker(salt=134305396, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=37061) done      HelloWorld()
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   HelloWorld__99914b932b   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Done
DEBUG: There are no more tasks to run at this time
INFO: Worker Worker(salt=134305396, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=37061) was stopped. Shutting down Keep-Alive thread
INFO:
===== Luigi Execution Summary =====

Scheduled 1 tasks of which:
* 1 ran successfully:
    - 1 HelloWorld()

This progress looks :) because there were no failed tasks or missing external dependencies

===== Luigi Execution Summary =====

まずはHelloWorldから。 末永くよろしくお願いいたします。