如何在Airflow中复用SnowflakeOperator执行多个.sql文件

核心思路是避免硬编码每个SQL对应的Operator实例，通过动态生成的方式批量创建任务，从根源上减少冗余代码。

方案1：动态遍历SQL列表生成独立任务（推荐） #

这种方案下每个SQL文件对应一个独立的Airflow任务，支持单独查看执行状态、配置重试、设置依赖关系，是生产环境最常用的方式。

• 首先统一维护SQL文件清单，既可以手动枚举路径，也可以通过脚本自动扫描指定目录下所有.sql后缀的文件，新增SQL时无需手动修改DAG代码
• 遍历清单动态实例化SnowflakeOperator，用SQL文件名自动生成唯一的task_id，避免重名
• 可以按需配置任务间的依赖关系，比如按文件名排序后串行执行，或者按业务逻辑分组并行

基础实现代码示例：

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow import DAG
from datetime import datetime
import os

# 基础DAG定义
my_dag = DAG(
    dag_id="snowflake_multi_sql_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False
)

# ---------- 两种SQL清单维护方式二选一 ----------
# 方式1：手动维护SQL路径列表
sql_file_list = [
    "/sql/abc.sql",
    "/sql/def.sql",
    "/sql/ghi.sql"
]

# 方式2：自动扫描指定目录下所有.sql文件，新增SQL无需改代码
# SQL_DIR = "/sql"
# sql_file_list = [
#     os.path.join(SQL_DIR, f)
#     for f in os.listdir(SQL_DIR)
#     if f.endswith(".sql")
# ]
# ------------------------------------------------

# 循环批量生成任务，如需串行执行可打开依赖配置逻辑
prev_task = None
for sql_path in sql_file_list:
    # 从SQL路径提取文件名生成唯一task_id，例如/sql/abc.sql对应task_id为sf_task_abc
    file_name = sql_path.split("/")[-1].replace(".sql", "")
    task_id = f"sf_task_{file_name}"
    
    current_task = SnowflakeOperator(
        task_id=task_id,
        dag=my_dag,
        sql=sql_path
        # 所有任务通用的参数直接在这里统一配置即可，比如连接ID、计算仓库、重试次数等
        # snowflake_conn_id="snowflake_default",
        # warehouse="ANALYSIS_WH",
        # retries=2
    )

    # 如果需要按顺序串行执行所有SQL，打开下面三行即可
    # if prev_task:
    #     prev_task >> current_task
    # prev_task = current_task

如果部分SQL需要特殊配置（比如用更大的计算仓库、更多重试次数），只需要把SQL清单改成字典格式，循环时读取对应配置传参即可，依然不需要重复写Operator实例化代码：

sql_file_list = [
    {"path": "/sql/abc.sql", "warehouse": "SMALL_WH", "retries":1},
    {"path": "/sql/heavy_etl.sql", "warehouse": "XLARGE_WH", "retries":3}
]

for sql_conf in sql_file_list:
    file_name = sql_conf["path"].split("/")[-1].replace(".sql", "")
    current_task = SnowflakeOperator(
        task_id=f"sf_task_{file_name}",
        dag=my_dag,
        sql=sql_conf["path"],
        warehouse=sql_conf["warehouse"],
        retries=sql_conf["retries"]
    )

方案2：合并多SQL为单个任务执行 #

如果不需要单独追踪每个SQL的执行状态，只是想批量跑一批强绑定的小SQL，可以在DAG解析阶段把所有SQL文件的内容读取拼接成一个完整脚本，传给单个SnowflakeOperator执行。

注意：该方案下所有SQL在同一个Snowflake会话中运行，任务失败时无法直接定位到具体哪个SQL文件报错，适合逻辑简单、不需要单独监控的轻量场景。

实现代码：

SQL_DIR = "/sql/small_batch_sqls"
full_sql = ""
# 按文件名排序保证执行顺序
for filename in sorted(os.listdir(SQL_DIR)):
    if filename.endswith(".sql"):
        with open(os.path.join(SQL_DIR, filename), "r", encoding="utf-8") as f:
            full_sql += f.read().strip(";") + ";\n"

batch_task = SnowflakeOperator(
    task_id="sf_batch_sql_task",
    dag=my_dag,
    sql=full_sql
)

注意事项 #

• 动态生成task_id时必须保证全局唯一，否则Airflow会抛出任务重复定义的错误
• 要确保Airflow的调度器、Worker进程能正常访问配置的SQL文件路径，否则运行时会报文件不存在的错误
• 如果用目录自动扫描的方式，注意不要把临时文件、备份文件（比如abc.sql.bak）放到目标目录下，避免被误识别成要执行的SQL

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