Airflow BigQueryInsertJobOperator重复执行问题排查与方案咨询

问题场景 #

我们的某一DAG存在如下问题：

• BigQueryInsertJobOperator任务因Airflow视角下的“僵尸任务”失败，事件日志显示心跳超时：

  Task did not emit heartbeat within time limit (300 seconds) and will be terminated. ...

• 该任务通过普通INSERT语句加载前一日数据，任务重试后，BQ中第一次尝试的任务仍在运行且几乎都会成功，最终导致数据重复插入。

尝试的解决方案及问题 #

我们尝试通过指定job_id参数实现幂等性：

• 根据Airflow文档，指定job_id后若存在同名任务则会重连至现有任务，因此设置了含日期变量的job_id及reattach_states=["RUNNING"]
• 实际发现：不同任务尝试的job_id后会附加UUID类后缀，导致无法实现重连。查看源码发现job_id会附加基于configuration的MD5后缀，但不清楚哪些配置变化导致不同尝试的后缀不同。

疑问 #

1. 是否误解了文档？正确解读是什么？
2. 解决该重复插入问题的最佳方案是什么？
3. 如何测试任务重连功能（如何创建同ID任务并验证Airflow重连）？

内容的提问来源于stack exchange，提问作者IgorM