Airflow中SnowflakeOperator执行赋值SQL匿名块报错求助

问题原因 #

报错显示SQL编译时遇到意外的EOF，从日志里的query: [BEGIN let env := current_database();]可以看出，Airflow的SnowflakeOperator默认会将SQL语句按分号拆分执行，导致匿名块被拆分成多个不完整的语句，第一个语句只执行了BEGIN let env := current_database();，缺少闭合的END;，因此触发语法错误。

解决方案 #

1. 禁用语句拆分（推荐）

在Airflow的SnowflakeOperator中添加split_statements=False参数，让Operator将整个匿名块作为单个SQL语句执行：

SnowflakeOperator(
    task_id='run_snowflake_anonymous_block',
    sql='path/to/your/sql/file.sql',
    snowflake_conn_id='your_snowflake_conn',
    split_statements=False,  # 关键参数，禁止拆分语句
    autocommit=True
)

2. 调整SQL文件格式

如果无法修改Operator参数，可以在SQL文件中将整个匿名块用单引号包裹（内部单引号需转义为两个单引号），确保被当作单个字符串处理：

'BEGIN
   let app := ''abc'';
   let env := current_database();
   let start_time := current_timestamp()::timestamp_ntz(9);
   let end_time := current_timestamp()::timestamp_ntz(9);
   let duration := DATEDIFF(seconds, start_time, end_time);  -- 原参数顺序会得到负时长，已修正
   let log_message := ''some log'';
   
   INSERT INTO proc_runtimes
      (env, app, task_name, start_time, end_time, duration, message)
   SELECT 
      :env, :app, ''Job Start'', :start_time, :end_time, :duration, :log_message;
END;'

3. 直接在Operator中定义SQL字符串

也可以直接在Operator的sql参数中定义完整的匿名块字符串，避免文件读取时的拆分问题：

SnowflakeOperator(
    task_id='run_snowflake_block',
    sql="""
    BEGIN
       let app := 'abc';
       let env := current_database();
       let start_time := current_timestamp()::timestamp_ntz(9);
       let end_time := current_timestamp()::timestamp_ntz(9);
       let duration := DATEDIFF(seconds, start_time, end_time);
       let log_message := 'some log';
       
       INSERT INTO proc_runtimes
          (env, app, task_name, start_time, end_time, duration, message)
       SELECT 
          :env, :app, 'Job Start', :start_time, :end_time, :duration, :log_message;
    END;
    """,
    snowflake_conn_id='your_snowflake_conn',
    split_statements=False,
    autocommit=True
)

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