問題タブ [luigi]

中央スケジューラ モードで luigi を使用しようとしています。コマンドを使用して中央スケジューラを実行しようとするたびに

構文エラーを返します

Luigi を使用していくつかのタスクを実行しています。次に、出力を標準化されたファイルの場所に一括転送する必要があります。complete()これを行うために、オーバーライドされたメソッドを使用して WrapperTask を作成しました。

complete()しかし、プロセスが実際に終了したときに呼び出される条件部分を取得するのに問題があります。

これは、他の人が指摘した非同期動作のためだと思いますが、修正方法がわかりません。

これらのコマンドライン パラメーターを使用して Luigi を実行してみました。

しかし、それは正しく機能していないようです。これは、このタイプのタスクを処理するための正しいアプローチですか?

また、私は興味があります — 誰かが--worker-retry-external-taskコマンドの経験がありますか? ちょっと理解に苦しむ。

ソースコードでは、

LuigiTask にメソッドがあるかどうかを判断するために呼び出されますが、run()メソッドはありWrapperTaskません。したがって、--retry-external-taskフラグがcomplete()完了するまでこれを再試行し、アクションを実行することを期待しています。ただし、インタプリタをいじってみると、次のように思われます。

このコード スニペットは、想定どおりの動作をしていません。

私はここで基地外ですか？

手動で作成されたファイルが存在するかどうかを確認する Luigi パイプラインに取り組んでおり、存在する場合は次のタスクに進みます。

私が望むのは、手動ファイルを作成してパスに貼り付けた後、ルイージが続行することです。これを行うと、ファイルを見つけてタスクを続行する代わりに、数秒ごとに新しいタスクを再チェックします。

かなりの時間 (15 ～ 20 分程度) が経過すると、luigi はファイルを見つけ、必要に応じて続行できるようになります。この遅延を防ぐにはどうすればよいですか? ファイルが存在し次第、ルイージに続行してもらいたい。

最近 LUIGI を使い始めましたが、ドキュメントを使用しても自分で答えられなかった質問がいくつかありました

質問は LUIGI のジョブの状態に関するものです
Luigi を使用すると、グローバル構成 (record_task_history) を設定してジョブの履歴を追跡でき、mysql DB をセットアップする必要があります。この履歴は後で取得して luigi ビジュアライザーで表示できます。

1. これに対して他の永続ストアはサポートされていますか? これはテキストファイルで実行できますか? (ビジュアライザーが機能しない可能性がありますが、プログラムで履歴状態を取得できれば問題ありません)

2. ルイージが実際に状態データを永続ストアに書き込むのはいつですか? ジョブを実行するためにメモリに保持し、ジョブが完了するとすぐにデータをダンプしますか?

--学習者

luigiを使用して etl パイプラインを作成しようとしています。ドキュメントから私が理解している限り、luigi のタスクは、ある種のファイル ストレージまたはデータベースのいずれかであるターゲットを生成できます。処理時間を短縮するには、メモリ内リストを出力したいと考えています。これは可能ですか？カスタム ターゲットを作成する必要がありますか?

データベースへのデータの更新/削除にluigiから使用できるモジュールは何ですか? データを挿入するためにテーブルへのコピーとSQL錬金術を使用しています。ドキュメントの更新と削除については、どのように達成できるか明確ではありませんか? お知らせ下さい。

私は docker swarm を使用して、機械学習アプリケーションに使用されるいくつかのコンテナーで展開を制御しています。

一部のコンテナーで実行されるコマンドを送信する bash スクリプトがあります。コンソールでこのスクリプトを手動で実行すると、すべてが正常に機能し、コマンドがコンテナーに送信されて実行されますが、この同じスクリプトを追加して cron で実行すると機能しません (私は crontab を使用しています - e コマンドを使用して、スクリプトをユーザーで実行できるようにします)

実行する必要があるスクリプトの一例を次に示します。

cron 行は次のとおりです。 16 * * * * sh /path/script.sh

swarm を使用しているため、最初に director の env にアクセスする必要があります。次に、docker exec を使用してコマンドを実行します (この場合は、spotify の luigi ツールを使用してジョブを開始するコマンドです)。

私は何が欠けていますか？ありがとう

requiresサブタスクであるルイージタスクがあります。サブタスクは、親タスク (つまり、requireing を実行しているタスク) によって渡されるパラメーターに依存します。サブタスクが使用できるパラメーターを設定することで指定できることは知っています...

...次に、サブタスクで、設定してパラメーターを受け取ります...

ただし、1 つのパラメーターのみを通過させるように見えます。どのようなタイプの任意の数のパラメーターを送信する最良の方法は何ですか? 本当に私はこのようなものが欲しいです:

ご覧のとおりluigi.DictParameter、ストレートの代わりに使用してみましたが、上記を実行するとルイージの奥深くからluigi.Parameter取得されます。TypeError: unhashable type: 'dict'

Python 2.7.11、ルイージ 2.1.1 の実行