訳あって、OSSのデータフロー制御ツールであるluigiで書かれたジョブのログを調べていたのだが、DEBUGとかの「そこは別にいいよ」というメッセージを結構大量に出す割に、実行したタスクがどれくらい時間がかかったかをログに出さないのである。

しかもログメッセージにデフォルトではタイムスタンプがついていない！

なんともなluigiさんだが、その後調べた結果、タスクの実行時間が取れる方法が分かったのでメモしておく。

環境

$ cat /etc/redhat-release 
CentOS Linux release 7.2.1511 (Core) 
$ python -V
Python 2.7.5

インストール

$ su -
# pip install luigi
# pip list | grep luigi
luigi (2.3.3)

デフォルトのログ

以下はサンプルのtop_artists.pyを動かした時に出力されるログメッセージの抜粋である。

$ PYTHONPATH='.' luigi --module top_artists AggregateArtists --local-scheduler --date-interval 2012-06
No handlers could be found for logger "luigi-interface"
DEBUG: Checking if AggregateArtists(date_interval=2012-06) is complete
DEBUG: Checking if Streams(date=2012-06-01) is complete
DEBUG: Checking if Streams(date=2012-06-02) is complete
（略）
INFO: Informed scheduler that task   AggregateArtists_2012_06_7af15faabf   has status   PENDING
INFO: Informed scheduler that task   Streams_2012_06_30_5c89db6ebe   has status   PENDING
INFO: Informed scheduler that task   Streams_2012_06_29_b5fcbcadc1   has status   PENDING
（略）
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 31
INFO: [pid 28321] Worker Worker(salt=588456947, workers=1, host=localhost.localdomain, username=hadoop, pid=28321) running   Streams(date=2012-06-06)
INFO: [pid 28321] Worker Worker(salt=588456947, workers=1, host=localhost.localdomain, username=hadoop, pid=28321) done      Streams(date=2012-06-06)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   Streams_2012_06_06_f7dba06bd7   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=588456947, workers=1, host=localhost.localdomain, username=hadoop, pid=28321) was stopped. Shutting down Keep-Alive thread
INFO: 
===== Luigi Execution Summary =====

Scheduled 31 tasks of which:
* 31 ran successfully:
    - 1 AggregateArtists(date_interval=2012-06)
    - 30 Streams(date=2012-06-01...2012-06-30)

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

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

実行時間の情報がない！！

Execution Summaryにいれとけよ！！(#^ω^)ビキビキ

タスクの実行時間を採る方法

luigiには内部で定義されたイベントが発生した際に呼びだされるコールバックを登録することができる。
その中に、Event.PROCESSING_TIMEという、タスクが正常終了した際に発行されるイベントがある。このイベントは内部にタスクがかかった時間を格納しているので、それを取り出すコールバックを登録しておけばよい。

すべて以下のリンク先の受け売りである。圧倒的感謝。

How to output the execution time of tasks in the luigi workflow system

コードも上のものを使って動作させると、以下のようなログメッセージになる。

実行に当たっては、カレントディレクトリにinput.txtを適当な内容で用意しておく。サンプルのコードはこのファイルを別名でコピーするタスクが実装されている。

$ cat input.txt
Hello Luigi!

実行。

$ python luigi_time_tasks_example.py --local-scheduler
DEBUG: Checking if TaskC() is complete
DEBUG: Checking if TaskB() is complete
INFO: Informed scheduler that task   TaskC__99914b932b   has status   PENDING
DEBUG: Checking if TaskA() is complete
INFO: Informed scheduler that task   TaskB__99914b932b   has status   PENDING
INFO: Informed scheduler that task   TaskA__99914b932b   has status   DONE
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 28797] Worker Worker(salt=876088965, workers=1, host=localhost.localdomain, username=hadoop, pid=28797) running   TaskB()
### PROCESSING TIME ###: 3.0078060627
INFO: [pid 28797] Worker Worker(salt=876088965, workers=1, host=localhost.localdomain, username=hadoop, pid=28797) done      TaskB()
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   TaskB__99914b932b   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 28797] Worker Worker(salt=876088965, workers=1, host=localhost.localdomain, username=hadoop, pid=28797) running   TaskC()
### PROCESSING TIME ###: 6.01015210152
INFO: [pid 28797] Worker Worker(salt=876088965, workers=1, host=localhost.localdomain, username=hadoop, pid=28797) done      TaskC()
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   TaskC__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=876088965, workers=1, host=localhost.localdomain, username=hadoop, pid=28797) was stopped. Shutting down Keep-Alive thread
INFO: 
===== Luigi Execution Summary =====

Scheduled 3 tasks of which:
* 1 present dependencies were encountered:
    - 1 TaskA()
* 2 ran successfully:
    - 1 TaskB()
    - 1 TaskC()

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

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

### PROCESSING TIME ###: （タスク実行にかかった時間[秒]）という形でログが追加出力されるようになる。

もう少し正確に述べると、リンク先のサンプルコードで

class TimeTaskMixin(object):
    '''
    A mixin that when added to a luigi task, will print out
    the tasks execution time to standard out, when the task is
    finished
    '''
    @luigi.Task.event_handler(luigi.Event.PROCESSING_TIME)
    def print_execution_time(self, processing_time):
        print('### PROCESSING TIME ###: ' + str(processing_time))

というようにluigiからPROCESSING_TIMEイベントに格納されてきた値をprintしている。

processing_timeに入っていくる値の単位が何かは上のサンプルだけだとさっぱりわからないが、luigi本体の実装を見ると秒数であることがわかる

自分で実装する時はどこかのファイルに書き出したり、DBに書いたりすればいいと思われる。

このPROCESSING_TIME機構だが、結構古いバージョンから実装されているようで、1.0.17あたりから入っている

以上。