Airflow中BigQueryInsertJobOperator无法建表且优先级异常求助

我正在配置Airflow任务，通过BigQueryInsertJobOperator执行BigQuery查询，期望目标表不存在时自动创建并存储查询结果。任务配置如下：

task3 = BigQueryInsertJobOperator(
             task_id="item_data",
             project_id="project_id",
             configuration={
    "jobType" : "QUERY",
    "query" : {
        "query" : "{% include 'sql_query.sql' %}",
        "useLegacySql" : False
    },
    "tableDefinitions" : {
        "fields" : [
            {
                "name" : "DEPT_NBR",
                "type" : "INTEGER"
            },
            {
                "name" : "ITEM_NBR",
                "type" : "INTEGER"
            },
            {
                "name" : "CREATED_DATE",
                "type" : "STRING"
            }
        ]
    },
    "destinationTable" : {
        "projectId" : "project_id",
        "datasetId" : "dataset_id",
        "tableId" : "table_id"
    },
    "createDisposition" : "CREATE_IF_NEEDED",
    "writeDisposition" : "WRITE_APPEND",
    "priority" : "BATCH",
    "schemaUpdateOptions" : [
        "ALLOW_FIELD_ADDITION"
    ],
    "timePartitioning" : {
        "type" : "DAY",
        "expirationMs" : 31556926000,
        "field" : "CREATED_DATE"
    },
    "clustering" : {
        "fields" : [
            "DEPT_NBR"
        ]
    }
},
             impersonation_chain="svc-account@project_id.iam.gserviceaccount.com",
             location="US" )

任务执行无报错，但出现两个异常：

1. 目标表未创建，数据被存储在24小时过期的临时表中
2. 已设置priority为BATCH，实际却以INTERACTIVE模式运行

一、目标表未创建、数据进入临时表的排查点 #

• 检查SQL查询逻辑：确认sql_query.sql中的语句是普通查询，没有包含CREATE TABLE AS SELECT、INSERT INTO等自行指定输出目标的语句。如果SQL已经显式定义了输出表，BigQuery会忽略配置中的destinationTable。
• 修正配置层级错误：QUERY类型任务中，destinationTable、createDisposition、writeDisposition等参数必须嵌套在query对象内部，而非顶层配置。当前顶层配置的这些参数无法被BigQuery识别，导致任务默认输出到临时表。正确结构示例：

  "query": {
      "query": "{% include 'sql_query.sql' %}",
      "useLegacySql": False,
      "destinationTable": {
          "projectId": "project_id",
          "datasetId": "dataset_id",
          "tableId": "table_id"
      },
      "createDisposition": "CREATE_IF_NEEDED",
      "writeDisposition": "WRITE_APPEND",
      # 其他相关参数也需移入此处
  }
  

• 移除多余的tableDefinitions配置：该参数用于定义外部数据源（如CSV、JSON）的结构，若你的查询未依赖外部表，此配置无意义，甚至可能干扰任务逻辑。
• 验证服务账号权限：确认svc-account@project_id.iam.gserviceaccount.com拥有目标数据集的bigquery.tables.create和bigquery.tables.updateData权限，避免因权限不足导致表创建失败。
• 检查目标表命名合法性：确认table_id符合BigQuery命名规则（无特殊字符、长度合规等），排除因命名不合法导致的创建失败。

二、优先级BATCH不生效的排查点 #

• 修正priority参数层级：和目标表配置一样，priority必须放在query对象内部，顶层配置的该参数无法被QUERY任务识别。
• 检查查询是否符合BATCH限制：BATCH优先级不支持部分交互式特性（如即时返回结果、特定临时表操作），若SQL包含这类特性，BigQuery会自动降级为INTERACTIVE模式。可在BigQuery控制台查看Job详情，确认是否存在优先级降级的提示。
• 升级Airflow Provider版本：旧版本的apache-airflow-providers-google包可能存在参数传递bug，导致priority配置不生效。尝试升级到最新稳定版后重新测试。

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