通过Airflow创建Snowflake存储过程遇截断问题，求解决方案

问题核心是Snowflake Operator默认会以分号为分隔符拆分SQL语句执行，但存储过程内部包含多个分号，导致语句被拆成不完整的片段，触发语法错误。以下是几种可行的解决方法：

方法1：设置split_statements=False参数 #

直接在Snowflake Operator的配置中添加该参数，让Operator将整个SQL语句作为单个单元执行，不再拆分。

示例代码：

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

create_proc_task = SnowflakeOperator(
    task_id="create_test_procedure",
    snowflake_conn_id="your_snowflake_conn",
    sql="""
CREATE OR REPLACE PROCEDURE test_proc() RETURNS VARCHAR NOT NULL LANGUAGE SQL AS
DECLARE
    exception_1 EXCEPTION (-20001, 'Table 1 failed');
    exception_2 EXCEPTION (-20002, 'Table 2 failed');
    exception_3 EXCEPTION (-20003, 'Table 3 failed');
BEGIN 
END;
    """,
    split_statements=False,  # 关键参数，禁止拆分SQL
)

方法2：用EXECUTE IMMEDIATE包裹存储过程语句 #

将整个存储过程创建语句作为字符串传入EXECUTE IMMEDIATE，这样整个SQL只有末尾一个分号，避免被拆分。注意用Snowflake的美元符号($$)作为字符串边界，无需转义内部单引号。

示例SQL：

EXECUTE IMMEDIATE $$
CREATE OR REPLACE PROCEDURE test_proc() RETURNS VARCHAR NOT NULL LANGUAGE SQL AS
DECLARE
    exception_1 EXCEPTION (-20001, 'Table 1 failed');
    exception_2 EXCEPTION (-20002, 'Table 2 failed');
    exception_3 EXCEPTION (-20003, 'Table 3 failed');
BEGIN 
END;
$$;

对应的Airflow Operator代码：

create_proc_task = SnowflakeOperator(
    task_id="create_test_procedure",
    snowflake_conn_id="your_snowflake_conn",
    sql="""
EXECUTE IMMEDIATE $$
CREATE OR REPLACE PROCEDURE test_proc() RETURNS VARCHAR NOT NULL LANGUAGE SQL AS
DECLARE
    exception_1 EXCEPTION (-20001, 'Table 1 failed');
    exception_2 EXCEPTION (-20002, 'Table 2 failed');
    exception_3 EXCEPTION (-20003, 'Table 3 failed');
BEGIN 
END;
$$;
    """,
)

方法3：从SQL文件加载语句并禁用拆分 #

如果存储过程逻辑复杂，建议将代码单独放在.sql文件中，通过Operator的sql参数引用文件路径，同时设置split_statements=False，便于后期维护。

比如创建create_test_proc.sql文件，内容为：

CREATE OR REPLACE PROCEDURE test_proc() RETURNS VARCHAR NOT NULL LANGUAGE SQL AS
DECLARE
    exception_1 EXCEPTION (-20001, 'Table 1 failed');
    exception_2 EXCEPTION (-20002, 'Table 2 failed');
    exception_3 EXCEPTION (-20003, 'Table 3 failed');
BEGIN 
END;

Airflow Operator配置：

create_proc_task = SnowflakeOperator(
    task_id="create_test_procedure",
    snowflake_conn_id="your_snowflake_conn",
    sql="path/to/create_test_proc.sql",
    split_statements=False,
)

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