ワークフローに Luigi を使用しています。私のワークフローは、インポート、分析、エクスポートの 3 つの一般的な部分に分かれています。各パーツ内には、複数のルイージ タスクがあります。

1 つのファイルにすべてを含めることができます。しかしdata_import.py、 、analysis.py、およびexport.py.

たとえば、data_import.py次のようになります。

しかし、export.py のタスクが import.py のタスクに依存している場合はどうでしょう。私はしますか：

大きなプロジェクトが複数のファイルに分割されている.py場合、Luigi に必要なタスクがどのファイルにあるかを伝える最良の方法は何ですか? この方法は面倒になりそうです。

失敗したタスクが数回再試行されるように、Luigi の再試行メカニズムを構成しようとしています。ただし、タスクは正常に再試行されますが、Luigi は失敗して終了します。

問題は、Luigi (pip install でバージョン 2.3.3 をインストールしました) をどのように構成すれば、タスクが一度失敗した後、成功して再試行されたときに、Luigi がThis progress looks :)で失敗する代わりに で正常に終了するようにするにはどうすればよいThis progress looks :(ですか?

これは、私が思いついた最小限のスケジューラとワーカー構成、および動作を示すためのタスクです。

mytasks.py:

実行例:

PYTHONPATH=. luigi --module mytasks MasterTask --workers=2 --path='/tmp/job-id-18'

したがって、質問はおそらくかなり単純です。~30 分間実行することになっているジョブがあり、プログラムが結果を 30 分待たされることは望ましくありません。タスク名や ID などを取得し、制御をユーザーに戻して、後でユーザーが自分のプログラムを使用してプロセスの状態を照会できるようにしたいと考えています。

https://github.com/spotify/luigi/blob/695010420105dc836096201a00d2f032f0143b84/luigi/interface.py

interface.py を見ると、関連するものは何もありません。それは不可能ですか？または、どうすればこれを達成できますか？ありがとう！

ps。luigi デーモンにタスクのステータスを照会する場所を誰かが教えてくれたら、ありがたいです。

S3 エンドポイントをシミュレートするためにFakeS3を使用して、vagrant マシン内でLuigiパイプラインをテストしようとしています。botoがFakeS3とやり取りできるようにするには、次のように接続をセットアップする必要があります。OrdinaryCallingFormat

しかし、Luigi を使用する場合、この接続はs3 モジュールに埋もれています。luigi.cfgmyを変更し、次のように s3 セクションを追加することで、ほとんどのオプションを渡すことができました

に必要なオブジェクトを渡す方法がわかりませんcalling_format。

今、私は立ち往生しており、先に進む方法がわかりません。私が考えることができるオプション：

1. OrdinaryCallingFormattoを渡す方法を理解S3Connectionするluigi.cfg
2. botoVagrant マシンで常にこの呼び出し形式を強制的に使用する方法を見つけて.aws/configください。boto.cfg
3. それがたまたま使用するFakeS3デフォルトを受け入れるようにします（それが何を意味するにせよ）。calling_formatbotoSubdomainCallingFormat

これを修正する方法についてのアイデアはありますか?

Luigi パイプラインの一部として、POST リクエストを使用して計算中のデータを待っているマイクロサービスに通知したいと考えています。

今までは を使ってRunAnywayTargetいましたが、データの変化率よりも早くルイージを起動させてしまうと困ります。だから私の質問は、

REST サービスへの POST リクエストの実行、Kafka へのメッセージの送信など、パイプラインで何かを実行するがデータの一部を作成しないタスクを作成するための最適なパターンは何ですか?

run メソッドで要求を実行する出力のないタスクを作成できることはわかっていますが、最初の実行中に何らかの理由でエンド サービスが失敗した場合、この NotificationTask を再実行するにはどうすればよいでしょうか? 依存関係が存在し、再度実行されることはありません。

Linux ターミナルと対話する Python3.4 Luigi パイプラインを使用しており、ユーザー入力を取得する必要があります。以下のようなコードがあります。

の後にターミナルが何も出力しないことを期待します"Enter data: "が、デバッグ情報でターミナルを更新し続けるため (つまり、スケジューラに作業を依頼し、クラス 1 が終了するのを保留中)、ユーザーは入力を入力できません。ユーザー入力を受け取るまで端末が印刷しないようにする方法はありますか?

ありがとう

アップデート

ユーザーが入力を入力しなくても、クラス One が実行されるようです。なぜこれが起こっているのかわかりません..

私の Windows VM では、Luigi でいくつかの小さなジョブを実行しています。すべての実行の開始時に「No Instance(s) Available」というメッセージが表示されるのはなぜだろうと思っています。プログラムを再実行するたびに、これらのうちの 1 つが出力されます。そのため、10回実行した後、「使用可能なインスタンスがありません」というメッセージが10回表示されます。

私が実行しているモジュールには影響しないようですので、好奇心から尋ねているだけです: Luigi/Command Prompt がこれを教えてくれるのはなぜですか? また、影響はありますか?

編集: getpcmd は、メッセージを生成する wmic コマンドのソースです。そのトレイルは次のとおりです。

犯人だと思うけど_WorkerScheduleFactory憶測だよ

私のチームはSentryエラーの追跡に使用しているため、Luigi の組み込みの電子メール機能を使用してすべてのレポートを 1 か所にまとめることは避けたいと考えています。

これは私が現在設定している方法であり、Sentryを完全にスキップしているようです: