問題タブ [airflow]

Airflow は主に ETL/入札データ関連のジョブに使用されています。ユーザーアクションが将来的に一連の依存タスクをトリガーするビジネスワークフローに使用しようとしています。これらのタスクの一部は、他の特定のユーザー アクションに基づいてクリア (削除) する必要がある場合があります。これを処理する最善の方法は、動的タスク ID を使用することだと思いました。Airflow が動的 DAG ID をサポートしていることを読みました。そこで、DAG ID とタスク ID をコマンド ライン パラメーターとして受け取る単純な Python スクリプトを作成しました。しかし、私はそれを機能させるのに問題が発生しています。dag_id が見つからないというエラーが発生します。誰もこれを試しましたか？コマンドラインでpythonとして実行するスクリプト（tmp.pyと呼びます）のコードは次のとおりです（python tmp.py 820 2016-08-24T22:50:00）：

気流とcronの実行は非常に混乱していると言わざるを得ません。今から5分ごとにcronを開始したいだけです。複雑な DAG の作成は簡単です。ドキュメントに基づいてcronを実行するロジックを理解することはそうではありません。

以下のコードを実行するとどうなりますか? 実行される毎秒、以下を印刷します。

以下は私のコードです。

データ パイプラインの気流を実験しています。残念ながら、これまでのところ、bigquery オペレーターで動作させることはできません。できる限りの解決策を探しましたが、まだ行き詰まっています.ローカルで実行されているシーケンシャルエグゼキューターを使用しています。

これが私のコードです：

エラーメッセージ:

Airflow インストールでPython Operator Exampleを実行しようとしました。インストールにより、Web サーバー、スケジューラ、およびワーカーが同じマシンに展開され、PytohnOperator 以外のすべてのタスクで問題なく実行されます。タスクは失敗し、モジュール「unusual_prefix_*」をインポートできませんでした。* は DAG を含むファイルの名前です。

完全なスタック トレース:

一定期間後、気流が特定のプールに定義されたスロット制限を超えます。再起動後は問題なく動作し、これに対する恒久的な解決策を探します。

気流バージョン - 1.6.2

前もって感謝します。

セロリエグゼキューターでエアフローを使用しています。現在、DAG で BranchOperator を使用してブランチの選択を決定するタスクのユーザー操作を開発することを計画しています。データベースの値をチェックするために連続ループを実行することで機能します。しかし、それは良い方法ではないと感じています。これを行う代替手段はありますか？

そして、私はこの相互作用を特定の時間まで待ちたいのですが、そうでなければ止めたいと思います. 気流でこれを行うことは可能ですか？可能であれば、この時間を動的に変更する可能性はありますか?

前もって感謝します。

pipなしでAirflowをインストールする方法はありますか?

pip を持たないオフライン コンピューターに Airflow をインストールしようとしています。インターネットからパッケージをダウンロードしましたが、pip なしでインストールを実行する方法がわかりません。

「setup.py」を使用してインストールを実行する方法を知っている人はいますか?