回调故障时的气流 [英] Airflow on_failure_callback
本文介绍了回调故障时的气流的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我的气流DAG有两个任务:
- Read_CSV
- 进程文件
他们自己工作得很好。我故意在 pandas 数据帧中创建了一个打字错误,以了解on_failure_callback
是如何工作的,并查看它是否被触发。从日志中看似乎并非如此:
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1197, in handle_failure
task.on_failure_callback(context)
TypeError: on_failure_callback() takes 0 positional arguments but 1 was given
on_failure_callback
为什么不工作?
以下是DAG的直观表示:
代码如下:
try:
from datetime import timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import pandas as pd
# Setting up Triggers
from airflow.utils.trigger_rule import TriggerRule
# for Getting Variables from airlfow
from airflow.models import Variable
print("All Dag modules are ok ......")
except Exception as e:
print("Error {} ".format(e))
def read_csv(**context):
data = [{"name":"Soumil","title":"Full Stack Software Engineer"}, { "name":"Nitin","title":"Full Stack Software Engineer"},]
df = pd.DataFramee(data=data)
dag_config = Variable.get("VAR1")
print("VAR 1 is : {} ".format(dag_config))
context['ti'].xcom_push(key='mykey', value=df)
def process_file(**context):
instance = context.get("ti").xcom_pull(key='mykey')
print(instance.head(2))
return "Process complete "
def on_failure_callback(**context):
print("Fail works ! ")
with DAG(dag_id="invoices_dag",
schedule_interval="@once",
default_args={
"owner": "airflow",
"start_date": datetime(2020, 11, 1),
"retries": 1,
"retry_delay": timedelta(minutes=1),
'on_failure_callback': on_failure_callback,
},
catchup=False) as dag:
read_csv = PythonOperator(
task_id="read_csv",
python_callable=read_csv,
op_kwargs={'filename': "Soumil.csv"},
provide_context=True
)
process_file = PythonOperator(
task_id="process_file",
python_callable=process_file,
provide_context=True
)
read_csv >> process_file
# ====================================Notes====================================
# all_success -> triggers when all tasks arecomplete
# one_success -> trigger when one task is complete
# all_done -> Trigger when all Tasks are Done
# all_failed -> Trigger when all task Failed
# one_failed -> one task is failed
# none_failed -> No Task Failed
# ==============================================================================
# ============================== Executor====================================
# There are Three main types of executor
# -> Sequential Executor run single task in linear fashion wih no parllelism default Dev
# -> Local Exector run each task in seperate process
# -> Celery Executor Run each worker node within multi node architecture Most scalable
# ===========================================================================
推荐答案
您需要为可以接收上下文的函数指定一个参数这是由于气流触发的方式on_failure_callback
def on_failure_callback(context):
print("Fail works ! ")
请注意,在您的实现中,您无法从消息中看出哪个任务失败了,因此您可能需要将任务详细信息添加到错误消息中,如下所示:
def on_failure_callback(context):
ti = context['task_instance']
print(f"task {ti.task_id } failed in dag { ti.dag_id } ")
这篇关于回调故障时的气流的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文