DAG障害のグローバルアラート

Question

現在、100個以上のDAGが稼動しています。 on_failure_callbackとアラートを追加する方法と、上流側の障害によってトリガーされたオペレータに気付いていますが、Airflow自体を構成して、DAGが失敗したときに電子メールを送信して、個別に失敗したら？

私の知る限りでは

Answer 1

ないが、私は私のグローバル/デフォルトのDAG /オペレータの設定を処理するために、このヘルパーを持っている：

def on_failure_callback(context): 
    ... 

def on_success_callback(context): 
    ... 

def build_default_args(**kwargs): 
    default_args = { 
     'on_failure_callback': on_failure_callback, 
     'on_success_callback': on_success_callback, 
     'owner': 'me', 
     'queue': 'default', 
     'execution_timeout': timedelta(hours=1), 
     'retries': 3, 
     'retry_delay': timedelta(seconds=10), 
    } 
    default_args.update(kwargs) 
    return default_args 

その後、各DAG内：

dag = DAG(
    dag_id='my_dag', 
    default_args=build_default_args(
     start_date=datetime(2017, 9, 20), 
     execution_timeout=timedelta(hours=8), # overrides default 
    ), 
    schedule_interval='@hourly', 
) 

また、いくつかのカスタムベースDAGクラス...どちらの方法でも、あなたは100回以上のDAGを一度変更して戻ってください。