如何基于BigQuery查询结果使Airflow任务按需失败？

要让任意检查返回1（或'False'）时触发Airflow任务失败，核心思路是在SQL脚本中检测到失败条件时主动抛出BigQuery异常——因为BigQueryInsertJobOperator会将BigQuery作业的失败直接映射为Airflow任务失败。

具体SQL改造方案 #

将多个检查查询合并，汇总结果后判断是否存在失败项，若存在则抛出带具体信息的异常：

-- 1. 汇总所有检查结果，给每个检查命名方便排查
WITH check_results AS (
  SELECT 
    '日活阈值检查' AS check_name,
    IF(last_week_count/7 > average_daily_threshold, 0, 1) AS check_result 
  FROM (SELECT ...) -- 你的第一个检查查询逻辑
  UNION ALL
  SELECT 
    '留存率阈值检查' AS check_name,
    IF(last_week_count/7 > average_daily_threshold, 0, 1) AS check_result 
  FROM (SELECT ...) -- 你的第二个检查查询逻辑
),
-- 2. 筛选出所有未通过的检查
failed_checks AS (
  SELECT check_name FROM check_results WHERE check_result = 1
)
-- 3. 若存在未通过检查，抛出异常终止作业
SELECT
  IF(EXISTS(SELECT 1 FROM failed_checks),
     RAISE_EXCEPTION('数据检查未通过：' || ARRAY_TO_STRING(ARRAY_AGG(check_name), '、')),
     0) AS check_status
FROM failed_checks;

关键细节调整 #

• 如果你的检查逻辑返回的是'False'字符串而非数字1，只需将check_result = 1替换为check_result = 'False'。
• 异常信息中包含未通过检查的名称，能快速定位问题，无需额外日志排查。

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