問題タブ [luigi]

LocalTargetプロジェクトの Luigi パイプラインでバイナリを作成するのに問題があります。ここで問題を切り分けました：

として開いてみまし'w'た'wb'が、次のエラーが引き続き発生します。

私はpython 3.5.1を使用しており、ルイージのバージョンは2.1.1です

Luigi Tasks のラッパーRegisterを作成していますが、実際には ABC メタクラスであり、 dynamic を作成するときに選択できないクラスで問題が発生しましたtype。

次のコードは、多かれ少なかれ、動的クラスを開発するために使用しているものです。

ただし、呼び出し関数を実行すると、PicklingError: Can't pickle <class 'abc.ScrubbedNameTask'>: attribute lookup abc.ScrubbedNameTask failed.

呼び出し機能:

ルイージがどのように機能するかを理解しようとしていますが、アイデアは得られますが、実際の実装は少し難しいです;)これは私が持っているものです:

そして、これは失敗しRuntimeError: Unfulfilled dependency at run time: OtherTask_3_5862334ee2ます。def output(self):この問題\機能を回避するには、を使用して出力を生成する必要があると考えました。そして、ファイルに書き込まずに適切な出力を生成する方法を理解できません。たとえば、次のように言います。

ドキュメントを読んでみましたが、出力の背後にある概念がまったくわかりません。画面のみに出力する必要がある場合はどうすればよいですか。オブジェクトを別のタスクに出力する必要がある場合はどうすればよいですか? ありがとう！

ルイージのタスクごとの再試行ポリシーの構成に問題があります。グローバル luigi.cfg ファイルを次のように構成しました。

さらに、ルイージの設定マニュアルには、タスクを次のように記述すると記載されています。

luigi.cfg で指定された luigi retry_count を上書きするだけで十分です。ただし、この設定は実行にはまったく影響しません。テストのためだけに毎回失敗するタスクを作成できましたが、ログには、このタスクが 5 回 (3 回ではなく) 失敗したことが返されます。

私が欠けている基本的な何かがあると思います。

Luigi で MongoDB に出力する方法があれば教えてください。ファイル（ローカルFS、HDFS）、S3、PostgreSQLをサポートしていますが、MongoDBはサポートしていません。そうでない場合、誰かが私に理由を説明できますか? 持っててもダメなのかな。データをデータベースに保存したいのは、クエリを実行してデータを探索できるからです。しかし、私は mongodb を使用しており、別のデータベースをインストールしたくありません。データベースを関係なしで保存およびクエリ ( NoSql ) にのみ使用しているため、リレーショナル データベースは必要ありません。したがって、最良のオプションは mongodb です。

基本的に、データを読み取ってデータベースに保存するタスクが必要です。次に、次のタスクがこの出力を取得してデータを処理します。

推奨事項、提案、または説明は大歓迎です。ありがとう！

現在、Ubuntu サーバーの電源が入ったときに Luigid を起動しようとしています。rc.local、cronjob(@reboot)、upstart、systemd などのいくつかの手法を試しましたが、どれも機能していないようです。

手動でコマンドを実行すると、コマンドが正常に実行されることを指摘する必要があります。起動時に実行する必要があるだけです。この時点で、どの方法で機能させるかについてはまったく心配していないので、私が試したことのいくつかを以下に示します-

クロン:

使用済み

そして入った

システムド:

/usr/bin に次の内容の luigid というスクリプトがあり、実行可能としてマークされています。適切な終了コードが必要になる可能性があることを恐れて、「exit 0」の有無にかかわらず試しました-

luigid.service という /etc/systemd/system/ のサービスファイル -

私は forking を試みました（そして、サービスと lugid コマンドの両方で PIDfile を指定しました。ワンショットで単純なタイプとして運がありませんでした。

を使用してサービスを有効にしました -

を使用してステータスを確認するため、サービスを開始しようとしているようです

systemctl status luigid.service

ショー

私が見逃している明らかな何かがあるに違いありません。起動時にコマンドを実行するのはそれほど難しいことではありません!

多くのプロジェクトで、パイプライン ツールとしてluigiを使用しています。これを利用してパラメータ検索を実装することを考えました。標準luigi.file.LocalTargetには、パラメーターを処理するための非常に単純なアプローチがあり、ドキュメントの例にも示されています。

つまり、パラメータはファイル名で保存されます。これにより、特定のパラメーターの組み合わせが既に計算されているかどうかを簡単に確認できます。これは、タスクのパラメーターがより複雑になるとすぐに厄介になります。

パラメータ検索の非常に単純なアイデアを次に示します。

確かに、この例では、4 つのパラメーターすべてをファイル名にエンコードできますが、このアプローチが境界に到達できるという多くの空想は必要ありません。配列のようなパラメーターの例を考えてみてください。

私のフォローアップのアイデアは、パラメータと結果をある種のエンベロープオブジェクトに保存し、それをターゲットとして保存することでした。ファイル名は、最初のあいまい検索のパラメーターのある種のハッシュである可能性があります。

封筒クラスあり

次に、LocalTarget を強化し、エンベロープ内のすべてのパラメーターが一致しているかどうかを確認できる新しいターゲットがあります。

ここでの問題は、このターゲットを使用すると、もともと luigi が最小化することを目指していたボイラープレートが追加されることです。新しい基本タスクを設定しました

結果のEnvelopedSumTask は非常に小さくなります。

Sumこのタスクは、最初のタスクと同じ方法で実行できます。

注: luigi-task-results をエンベロープする方法のこの実装例は、安定とはほど遠いものであり、結果とパラメーターをエンベロープすることによって私が意味するものをより具体的に示しています。

私の質問は: ルイージで多くの複雑なパラメーターを処理する簡単な方法はありませんか?

フォローアップの質問: パラメータ検索が実行されたコード バージョン (および/またはサブタスクのパッケージ バージョン) の記録を保持することを考えた人はいますか?

このトピックのどこを読むべきかについてのコメントも大歓迎です。

ノート：

これを実行するには、おそらくいくつかのインポートが必要です。

プライマリ データ ストア (および「Universal Source of Truth」) が Postgres であるシステムがありますが、そのデータをリアルタイムと夜間の両方で集約してレプリケートします。現在、Elasticsearch、Redis、Redshift (夜間のみ) にレプリケートしており、Neo4j も追加しています。

私たちの ETL パイプラインは十分に拡張されており、 AirflowやLuigiなどのツールを検討し始めていますが、最初の調査からわかることから、これらのツールはほぼ完全に一括読み込みを目的としています。

大規模なバッチ ETL プロセスと、オンザフライで大量の個別レコードのレプリケーションの両方を処理できる ETL プロセスを処理できるツールはありますか? Airflow または Luigi はこれを処理しますか?

ありがとう！