問題タブ [airflow]

コード：

Python バージョン 2.7.x および airflow バージョン 1.5.1

私のdagスクリプトはこれです

そこから、6 つのタスクで DAG を作成していることがわかります。最初のタスク (Start1) が最初に開始され、その後、他の 5 つのタスクがすべて開始されます。

現在、DAG の開始の間に 5 分の時間遅延を与えています

最初のタイプの 6 つのタスクすべてで完全に実行されましたが、5 分後に DAG が再開されません。

1 時間以上経過しても、DAG が再起動されません。私が間違っていたかどうかはわかりません。

誰かが私に何が間違っているのかを指摘できれば、それは本当に素晴らしいことです.私airflow testing clearは同じことが起こるために使用してクリアしようとしました.それは最初のインスタンスを実行した後、ただそこに立っていました.

コマンドラインに表示される唯一のものはGetting all instance for DAG testing

schedule_interval の位置を変更すると、スケジュール間隔なしで並列に実行されます。つまり、5 分で 300 以上のタスク インスタンスが完了します。5分のスケジュール間隔はありません

コード 2:

職場では、AirBnB による Airflow の PoC を完了しています。標準的な運用方法は、開発 (つまり、DAG の作成や Python スクリプトなど) をランタイム環境から切り離すことです。つまり、サーバー上で直接スクリプトを作成することは許可されていません (とにかく悪い習慣です)。

そのため、Windows で Python (期待どおりに動作) で動作するように intelliJ を構成しました (UNIX ワークステーションを使用することは許可されていません) が、Windows にエアフローをインストールする方法が見つかりません。Java 開発のバックグラウンドを持つ Python は初めてです。

そのため、AirBnb/Airflow 用のローカル開発環境 IDEA Ultimate をセットアップするにはどうすればよいですか (ドキュメントはありませんか?

Airflow を使用して単純なタスク python を実行しようとしています。

たとえば、試してみると：

気流テスト python_test print 2015-01-01

できます！

def print_context(ds, **kwargs)今、私は自分の関数を他のpythonファイルに入れたいと思っています。だから私はという別のファイルを作成します: simple_test.py と変更:

今、私は再び実行しようとします:

気流テスト python_test print 2015-01-01

そしてOK！それはまだ動作します！

ただし、ファイルを使用してワーカー モジュールなどのモジュールを作成する場合は、それをSimplePython.pyインポート ( from worker import SimplePython) して試してください。

気流テスト python_test print 2015-01-01

次のメッセージが表示されます。

ImportError: ワーカーという名前のモジュールがありません

質問：

1. DAG 定義内にモジュールをインポートすることはできますか?
2. Airflow+Celery は、必要なすべての Python ソース ファイルをワーカー ノード間でどのように配布しますか?

気流でハイブオペレーターを使用しようとしています。すべての依存関係がインストールされています (pyhs2、pyhive、および pip install airflow[hive] を実行しました)。

ただし、以下のコードを使用すると

このエラーが発生します。私はそれが何を意味するのか分かりません

私は airbnb の気流を使用しており、以下に示す簡単なタスクを作成しています。ただし、間隔を毎時またはその他の間隔に設定しても、スケジューラはタスクを実行し続けます。私が気付いたもう 1 つのことは、スケジュール間隔を「@once」に設定すると、dag が実行されないことです。

ここの規則に従いましたhttp://airflow.readthedocs.org/en/latest/scheduler.html#dag-runs

私が使っているシンプルなダグ。

参照用のスケジューラ出力。ご覧のとおり、何度も実行し続けていますが、私の日記では、schedule_interval='@hourly' があります。

特定の EmailOperator タスクによって送信される電子メールの内容を動的に設定できるようにする方法を探しています。理想的には、できれば html_content 引数を使用して、xcom 呼び出しの結果に依存する電子メールのコンテンツを作成したいと考えています。

気流のドキュメントには、xcom 呼び出しをテンプレートに埋め込むことができると書かれていることに気付きました。おそらく、指定されたタスク ID のテンプレートを使用して xcom プルを作成し、結果を html_content として渡す方法はありますか? ありがとう

airbnb airflowを試しています。「バックフィル」オプションを使用して、timedelta を 60 分に設定して 1 日実行しようとしていますが、13 個のインスタンスしか実行されません。残りは待機中として表示され、実行されません。