如何在Snowflake的Snowpark环境中部署并使用第三方库PandasAI

核心问题分析 #

• 代码直接将Snowpark DataFrame传入SmartDataframe，但SmartDataframe仅支持本地Pandas DataFrame，Snowpark DataFrame是分布式延迟执行对象，无法直接兼容。
• 未正确配置第三方库的依赖加载逻辑，Snowpark需要显式从Stage加载自定义库及其依赖包。

分步解决方案 #

1. 上传所有依赖库至Snowflake Stage #

PandasAI依赖pandas、numpy等基础库，需将所有依赖的.whl文件或打包zip上传至Stage：

-- 创建专属Stage（如不存在）
CREATE STAGE IF NOT EXISTS PANDASAI_LIBS;

-- 本地文件上传至Stage（替换为实际本地文件路径）
PUT file:///本地路径/pandasai-*.whl @PANDASAI_LIBS AUTO_COMPRESS=FALSE;
PUT file:///本地路径/pandas-*.whl @PANDASAI_LIBS AUTO_COMPRESS=FALSE;
PUT file:///本地路径/numpy-*.whl @PANDASAI_LIBS AUTO_COMPRESS=FALSE;

2. 在Snowpark会话中加载依赖库 #

在代码开头添加库加载逻辑，确保Snowpark能访问Stage中的依赖：

import snowflake.snowpark as snowpark
import sys

def main(session: snowpark.Session):
    # 加载Stage中的第三方库
    session.add_import("@PANDASAI_LIBS/pandasai-*.whl")
    session.add_import("@PANDASAI_LIBS/pandas-*.whl")
    session.add_import("@PANDASAI_LIBS/numpy-*.whl")

    # 设置Python路径，让解释器识别加载的库
    sys.path.append('/tmp')
    sys.path.append('/tmp/snowflake/snowpark')

3. 修正代码逻辑：转换为Pandas DataFrame #

将Snowpark DataFrame转换为本地Pandas DataFrame后，再传入SmartDataframe：

# 后续逻辑放在main函数内
    # 创建表并插入数据
    session.sql("CREATE TABLE IF NOT EXISTS my_table3 (STATUS varchar(255))").collect()
    session.sql("INSERT INTO my_table3 (STATUS) VALUES ('Fantastic')").collect()

    # 查询并转换为本地Pandas DataFrame
    snowpark_df = session.sql("SELECT STATUS FROM my_table3")
    pandas_df = snowpark_df.to_pandas()

    # 初始化SmartDataframe并执行查询
    from pandasai import SmartDataframe
    smart_dataframe = SmartDataframe(pandas_df)
    filtered_dataframe = smart_dataframe.query("STATUS == 'Fantastic'")
    
    print(filtered_dataframe)
    snowpark_df.show()

    return snowpark_df

4. 关键注意事项 #

• 数据量限制：to_pandas()会将数据拉取到本地，仅适合小数据集；大数据集建议优先使用Snowpark原生操作，避免性能瓶颈。
• 权限配置：确保执行代码的角色拥有Stage的USAGE和READ权限。
• 版本兼容：上传的库版本需与Snowflake运行时的Python版本匹配（当前Snowflake支持Python 3.8/3.9等版本）。

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