問題タブ [luigi]

Table-xxxx-xx-xx日次一時テーブル ( ) を既存のテーブル ( )に追加することのみを目的とした Bigquery タスクがありますPersistingTable。

メソッドの処理方法がわかりませんoutput(self)。実際、プロセスが開始される前にすでに存在しているため、PersistingTableとして出力することはできません。luigi.contrib.bigquery.BigQueryTarget誰かが自分自身にそのような質問をしたことがありますか?

EMR Spark クラスター (YARN を使用) を実行しており、EMR マスターから直接 Luigi タスクを実行しています。S3 のデータに依存する一連のジョブがあり、いくつかの SparkSubmitTasks の後、最終的に Redshift になります。

私は2つの主な問題に直面しています:

1) ときどき、luigi は SparkSubmitTask がいつ完了したかを判断できないことがあります。ジョブが完了したと判断できません。

2) 何らかの理由で SparkSubmitTasks を実行でき、上に配置したタスクが Spark ジョブを終了した場合、出力タスクは実行されず、マーカー テーブルは作成も入力もされません。ただし、実際のテーブルは、実行される Spark ジョブで作成されます。RedshiftTarget の呼び出し方法を誤解していますか?

それまでの間、私はソースコードに慣れようとしています。

ありがとう！

私は Python の初心者です。RHEL Linux に Luigi-2.0.1 をインストールしました。サンプルプログラムを実行してみる

次のコマンドを実行しました

次のエラーが表示されます。

Web を調べたところ、なぜこのようなことが起こったのかわかりませんでした。私はpython 2.6を持っています。luigi は 2.6 と互換性がなく、python 3.0 以降でしか動作しないということですか?

どんな助けにも感謝します

現在、単純な依存関係 chain( ) を使用して、一連の luigi タスクを一緒にキューに入れていa -> b -> c -> dます。d最初に実行さaれ、最後に実行されます。aトリガーされるタスクです。

を除くすべてのターゲットはオブジェクトをa返し、文字列 (日付と時刻を含む) であるluigi.LocalTarget()単一のジェネリックを持ちます。luigi.Parameter()luigi 中央サーバー (履歴が有効になっている) で実行されます。

問題は、上記のタスクを再実行するとa、ルイージが履歴をチェックし、その特定のタスクが以前に実行されたかどうかを確認し、ステータスが DONE であった場合、タスクを実行しないことです (dこの場合)。文字列を変更しても役に立ちません (ランダムなマイクロ秒が追加されました)。タスクを強制実行するにはどうすればよいですか?

luigi タスクのエイリアスを作成しようとしています。通常、それを呼び出すには、次のように入力しますluigi --module myTask --parameters cats

ただし、エイリアスを作成して に短縮したいmyTask --parameters catsので、に行って.bash_profileこれを入れましたalias myTask='luigi --module myTask "$@"'

ただし、エイリアスを使用してタスクを再度呼び出そうとすると、次のエラーが発生します: luigi.task_register.TaskClassNotFoundException: No task myTask. 候補は次のとおりです: Config,ExternalTask​​,RangeBase,RangeDaily,RangeDailyBase,RangeHourly,RangeHourlyBase,Task,WrapperTask,core,execution_summary,retcode,scheduler,worker

.bash_profileこのエイリアスを機能させるために何か不足していますか?