エアフローダグの途中でタスクを途中で終了する方法はありますか？

Question

FTPサーバー上のファイルをチェックするダグがあります（エアフローは別のサーバーで実行されます）。ファイルが存在する場合、ファイルはS3に移動します（ここでアーカイブします）。そこから、ファイル名がSparkサブミットジョブに渡されます。スパークジョブは、S3（異なるサーバー上のスパーククラスタ）を介してファイルを処理します。私は複数のダグを持つ必要があるかどうかはわかりませんが、ここではその流れがあります。私が探しているのは、S3バケットにファイルが存在する場合にのみスパークジョブを実行することです。

S3センサーを使用しようとしましたが、タイムアウト基準を満たした後にタイムアウトに失敗するため、ダグ全体が失敗に設定されます。

check_for_ftp_files -> move_files_to_s3 -> submit_job_to_spark -> archive_file_once_done 

私は唯一つまたは複数のファイルをS3に移動されたONLY FTPチェックを行うスクリプトの後、すべてを実行します。

Answer 1

2つの異なるDAGを持つことができます。 1人はS3センサーしか持っておらず、5分ごとに走り続けます。ファイルを検出すると、2番目のDAGがトリガーされます。 2番目のDAGはファイルをS3に提出し、完了したらアーカイブします。最初のDAGでTriggerDagRunOperatorを使用してトリガすることができます。

Answer 2

彼が与えた答えはうまくいくでしょう。 もう1つのオプションは、Sensorが持つ "soft_fail"パラメータ（BaseSensorOperatorのパラメータ）を使用しています。このパラメータをTrueに設定すると、タスクが失敗するのではなく、タスクがスキップされ、ブランチ内のすべての後続タスクもスキップされます。

詳細はairflow codeを参照してください。