能否在Snowflake环境下构建不依赖Snowpark的DBT Python模型？

DevOps 小满

2026-6-29

问题

我有一条数据流水线，其中包含大量写入Snowflake的DBT Python模型，这些模型原本是Databricks中的工作簿，近期已全部迁移至Snowflake。原代码通过循环生成异步数据库插入操作，但Snowpark Session库并非线程安全，导致现在插入只能同步执行。请问是否可以构建脱离Snowpark运行的DBT Python模型，从而使用Snowflake Python Connector进行异步数据库调用，而非强制使用Snowpark Session库？

原Databricks异步执行代码如下：

pool = ThreadPool(5)
pool.map(
    lambda statement: execute_statement(statement), statements_list)

解决方案

完全可以构建脱离Snowpark的DBT Python模型，改用Snowflake Python Connector实现异步插入，具体操作如下：

禁用Snowpark依赖：在Python模型中不要引入Snowpark Session，直接通过Snowflake Python Connector建立连接。可以利用DBT的环境变量或配置参数获取Snowflake的账户、仓库、数据库、Schema、用户名、密码等连接信息，全程无需依赖Snowpark的session对象。
线程池异步执行逻辑：Snowflake Python Connector的连接对象同样非线程安全，因此每个线程必须单独初始化连接。基于此调整异步执行逻辑，示例代码如下：

from snowflake.connector import connect
from multiprocessing.pool import ThreadPool
import os

def execute_statement(statement):
    # 每个线程独立创建连接
    conn = connect(
        account=os.getenv('SNOWFLAKE_ACCOUNT'),
        user=os.getenv('SNOWFLAKE_USER'),
        password=os.getenv('SNOWFLAKE_PASSWORD'),
        warehouse=os.getenv('SNOWFLAKE_WAREHOUSE'),
        database=os.getenv('SNOWFLAKE_DATABASE'),
        schema=os.getenv('SNOWFLAKE_SCHEMA')
    )
    cursor = conn.cursor()
    try:
        cursor.execute(statement)
        conn.commit()
    finally:
        cursor.close()
        conn.close()

def model(dbt, session):
    # 忽略Snowpark session参数，不依赖其功能
    dbt.config(materialized='table')
    
    # 生成待执行的SQL语句列表，逻辑与原代码一致
    statements_list = [...]
    
    # 线程池异步执行插入
    pool = ThreadPool(5)
    pool.map(execute_statement, statements_list)
    
    # 返回符合DBT要求的空DataFrame（若无需输出结果）
    return session.create_dataframe([], schema=[])

关键注意事项：
- 每个线程必须创建独立的Connector连接：共享连接会引发线程安全问题，导致执行异常。
- 通过DBT配置传递敏感信息：避免硬编码账户、密码等内容，可通过dbt.yml的vars或系统环境变量传递。
- 合理控制线程数：不要设置过高的线程池大小，防止触发Snowflake的并发请求限制，影响执行效率甚至导致请求被拦截。

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