Airflow BigQuery DAG突发ARRAY_LENGTH报错及PST 3点调度配置咨询

一、ARRAY_LENGTH类型报错根因定位 #

该类无代码变更触发的运行时类型错误，90%以上是数据侧变化触发了之前未覆盖的执行路径，常见原因按概率排序：

• 动态分支返回类型不统一：SQL中第70行传给ARRAY_LENGTH的表达式大概率是CASE WHEN/IF多分支生成的动态值，前2-3天的分区数据全部命中返回ARRAY类型的分支，未触发类型校验；当日数据命中了之前从未走到的分支，该分支返回了STRING类型值（比如空字符串''、逗号分隔的字符串字段、JSON_VALUE提取的字符串结果），直接触发类型不匹配报错。
• 上游表Schema无通知变更：你依赖的上游表（含视图、外部表）由其他团队/同步任务维护时，管理员可能将原本的ARRAY类型字段修改为STRING存储（比如将数组标签改为逗号分隔字符串存），未同步通知下游，读取时自然拿到STRING类型值。
• 动态SQL拼接转义失效：如果第70行的ARRAY对象是通过字符串拼接生成的，前几日传入的动态参数无特殊字符，拼接出的数组构造语法合法；当日参数带未转义的单引号、特殊符号，导致数组构造语法被截断，最终传入ARRAY_LENGTH的是被截断后的字符串字面量。

二、快速排查步骤 #

• 不要直接查源码里的模板SQL，去Airflow UI对应失败任务实例的Rendered Template页面，拿到Airflow渲染完成后实际发给BigQuery的完整SQL，定位第70行ARRAY_LENGTH的入参表达式。
• 复制该表达式，在BigQuery控制台绑定和失败任务完全一致的执行日期、分区过滤条件，执行SELECT TYPEOF(<你的入参表达式>)，确认返回类型是否为STRING。
• 溯源表达式来源：如果是直接读取的上游字段，查上游表近3天的Schema变更记录，确认字段类型是否改动；如果是多分支计算生成的值，逐分支检查返回值类型，确认是否存在返回STRING的分支；如果是拼接生成的SQL片段，检查当日传入的拼接参数是否存在未转义的特殊字符。

三、修复方案 #

• 加类型安全防护：对ARRAY_LENGTH的入参增加显式类型转换，或加安全前缀避免任务硬失败，示例：

  -- 如果入参可能是JSON格式的数组字符串，先转成ARRAY
  ARRAY_LENGTH(PARSE_ARRAY(your_param))
  -- 类型不匹配时返回NULL而非直接报错
  SAFE.ARRAY_LENGTH(your_param)
  

• 统一动态分支返回类型：所有CASE WHEN/IF分支的返回值必须为同一ARRAY类型，禁止某分支返回空字符串、字符串字段，空值分支要返回类型明确的空数组，示例：

  CASE
    WHEN xxx THEN SPLIT(tag_str, ',')
    -- 错误写法：ELSE ''
    ELSE CAST([] AS ARRAY<STRING>) -- 正确写法，类型和其他分支一致
  END
  

• 增加前置Schema校验：在DAG最开头增加校验节点，每次运行前先检查依赖上游表的字段类型是否符合预期，不符合则提前告警，避免任务跑到中途报错。

四、PST时区每日凌晨3点调度配置方法 #

直接在DAG初始化时绑定PST对应时区（自动处理夏令时切换，不要手动换算UTC时间写cron，避免夏令时偏移），配置示例：

from pendulum import timezone
import datetime
from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

# 绑定PST时区（America/Los_Angeles为PST/PDT对应的标准时区ID，自动适配夏令时）
pst_tz = timezone("America/Los_Angeles")

default_args = {
    "owner": "data_quality",
    "retries": 1
}

with DAG(
    dag_id="bq_data_quality_check",
    default_args=default_args,
    # 开始日期也要指定对应时区
    start_date=datetime.datetime(2024, 1, 1, tzinfo=pst_tz),
    # cron表达式：对应绑定时区下的每日3点整
    schedule="0 3 * * *",
    catchup=False,
    tags=["data_quality"]
) as dag:
    # 你的BigQuery校验任务逻辑
    check_job = BigQueryInsertJobOperator(
        task_id="run_quality_check",
        configuration={
            "query": {
                "query": "your_sql_path.sql",
                "useLegacySql": False
            }
        }
    )

配置完成后去Airflow UI查看DAG的Next Run时间，换算为PST时间确认是凌晨3点即可，Airflow会自动按照该时区的时间触发调度，无需手动调整时间偏移。

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