Airflow提取URL参数任务报错“Variable is not defined”求助

根据你提供的SQL代码和报错信息，错误根源在于Airflow变量替换机制与SQL引擎内置变量语法的冲突，具体分析如下：

核心问题点 #

你设置了variable_substitution=True，这个参数会让Airflow自动解析SQL中以&开头的内容，将其视为Airflow变量进行替换。但代码中的&__rowcount并非Airflow定义的变量，Airflow找不到对应值，因此抛出Variable is not defined错误。

另外，你可能混淆了SQL引擎（比如Snowflake）的内置行计数变量语法：多数SQL引擎中，执行DML后的行计数会话变量是$ROWCOUNT（大写），而非&__rowcount这种写法。

具体解决办法 #

方案1：关闭Airflow变量替换（推荐，若无需Airflow变量） #

如果你的SQL中不需要使用Airflow变量，直接关闭variable_substitution：

!set variable_substitution=False

同时修正行计数变量的语法，改用SQL引擎的内置会话变量（以Snowflake为例）：

SET row_count = $ROWCOUNT;

方案2：转义&符号，避免Airflow替换 #

如果必须保留variable_substitution=True，需要对&进行转义，让Airflow跳过解析，直接传递给SQL引擎：

SET row_count = \&__rowcount;

注：转义符可能因SQL引擎和Airflow执行器略有不同，若\无效可尝试&&。

方案3：改用Airflow获取行计数 #

通过Airflow的钩子（比如SnowflakeHook）执行CREATE TABLE语句后，直接获取返回的行计数，再作为变量传入后续的INSERT语句。示例伪代码：

from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook

def get_row_count():
    hook = SnowflakeHook(snowflake_conn_id='your_conn_id')
    sql = "CREATE OR REPLACE TABLE marketing_digital_traffic AS SELECT ..."
    result = hook.run(sql, return_last=True)
    row_count = result.rowcount
    return row_count

# 在Airflow任务中调用该函数，将row_count传入后续SQL

内容的提问来源于stack exchange，提问作者Joe F.