Pythonでのイベント処理Luigi

Question

私は、ワークフローハンドラとしてLuigiを統合しようとしてきました。現在はコンコースを利用していますが、私たちがやろうとしていることの多くは、コンコースで回るのが面倒なので、ルイージを依存マネージャーに切り替えました。これまでのところ問題はなく、ワークフローが適切に起動して実行されます。

タスクは何らかの理由でタスクが失敗したときに発生します。この場合、特にタスクのブロックが必要ですが、すべてのケースを考慮する必要があります。現時点では、Luigiはエラーを正常に処理し、それをSTDOUTに書き込みます。それでも、コード0を発行して終了しますが、コンコースにはジョブが渡されたことを意味します。偽陽性

も、非常に簡単な仕事で、私はこの問題を解決するためにイベント処理を取得しようとしてきたが、私はそれをトリガすることができません ：

@luigi.Task.event_handler(luigi.Event.FAILURE) 
def mourn_failure(task, exception): 
    with open('/root/luigi', 'a') as f: 
     f.write("we got the exception!") #testing in concourse image 
    sys.exit(luigi.retcodes.retcode().unhandled_exception) 

class Test(luigi.Task): 
    def requires(self): 
     raise Exception() 
     return [] 

    def run(self): 
     pass 

    def output(self): 
     return [] 

はその後のpythonシェルでコマンドを実行している

luigi.run(main_task_cls=Test, local_scheduler=True)

例外は発生しますが、それでも発生しません。 それは

[retcode] 
already_running=10 
missing_data=20 
not_run=25 
task_failed=30 
scheduling_error=35 
unhandled_exception=40 

が含まれています、私は/etc/luigi/client.cfgで私のルイージの設定を持っている違いがあれば、また、ファイルが書き込まれません。また、終了コードがまだ0

です私はイベントハンドラがなぜトリガされないのかについては迷っていますが、何とかエラーが発生した場合にプロセスが失敗する必要があります。

Answer 1

「例外発生」呼び出しを配置する場所が問題だと思われます。 require関数に配置すると、基本的にテストタスクの実行メソッドの前に実行されます。つまり、あなたのテストタスクが失敗したのではなく、それが依存しているタスク（今は空...）です。

たとえば、raiseを実行すると、コードは期待どおりに動作します。 （この場合には、例外がメソッドを必要に提起された）

def run(self): 
     print('start') 
     raise Exception() 

あなたの依存関係に障害が発生したケースを処理するために、あなたはルイージイベントハンドラ、BROKEN_TASKの別のタイプを追加することができます：luigi.Event.BROKEN_TASKを。 これは、luigiコードがあなたが期待する戻りコード（0以外）を発行することを保証します。

乾杯！

Answer 2

私が正しく理解していれば、タスクが失敗したときにluigiにエラーコードを返すようにしたいのですが、これには多くの問題がありましたが、それはかなりシンプルになります。ただluigi Pythonではなく、コマンドライン。このように：それはあまりにもあなたの問題だった場合

luigi --module my_module MyTask

私は知らないが、私は、Pythonで実行し、その後、ルイージはluigi.cfgにretcodesを無視しました。それが役に立てば幸い。