如何在Snowpark中读取Azure Blob存储的CSV文件为Snowpark DataFrame（无需加载至内存）

你遇到的问题其实是因为Snowpark和Spark的外部数据访问逻辑不一样——Spark可以直接调用云存储客户端读取文件，但Snowpark是依托Snowflake的**外部阶段（External Stage）**来访问外部存储的，不能直接用wasbs://路径去读。下面是完整的解决方案，完全不需要把数据加载到本地内存：

步骤1：在Snowflake中创建指向Azure Blob的外部阶段 #

首先需要把你的Azure Blob存储路径映射成Snowflake的外部阶段，同时配置访问凭证（比如SAS Token）和CSV文件格式。你可以直接用Snowpark执行SQL来创建：

# 先确保你已经建立了Snowpark Session（这里给个示例配置）
from snowflake.snowpark import Session

connection_params = {
    "account": "你的Snowflake账户名",
    "user": "你的用户名",
    "password": "你的密码",
    "warehouse": "使用的计算仓库",
    "database": "目标数据库",
    "schema": "目标Schema"
}
session = Session.builder.configs(connection_params).create()

# 创建CSV文件格式（如果还没有的话）
session.sql("""
CREATE OR REPLACE FILE_FORMAT csv_file_format
TYPE = CSV
FIELD_DELIMITER = ','
SKIP_HEADER = 1  # 适配你的CSV表头设置
FIELD_OPTIONALLY_ENCLOSED_BY = '"'
TRIM_SPACE = TRUE
""").collect()

# 创建指向Azure Blob的外部阶段
session.sql("""
CREATE OR REPLACE STAGE azure_blob_csv_stage
URL = 'wasbs://<你的容器名>@<存储账户名>.blob.core.windows.net/<CSV文件所在的路径>'
CREDENTIALS = (AZURE_SAS_TOKEN = '你的Blob SAS Token')
FILE_FORMAT = csv_file_format;
""").collect()

步骤2：读取CSV为Snowpark DataFrame #

现在你可以通过这个外部阶段来读取CSV了，解决Schema的问题有两种方式：

方式A：自动推断Schema（适合不知道字段结构的情况）

Snowflake提供了INFER_SCHEMA函数，可以自动从CSV文件中推断字段类型，然后转换成Snowpark的Schema：

from snowflake.snowpark.types import DataType

# 用INFER_SCHEMA获取CSV的Schema信息
schema_result = session.sql("""
SELECT INFER_SCHEMA(
    LOCATION => '@azure_blob_csv_stage',
    FILE_FORMAT => 'csv_file_format'
)
""").collect()

# 转换为Snowpark可识别的StructType
auto_schema = DataType.from_json(schema_result[0][0])

# 读取CSV文件
product_df = session.read.schema(auto_schema).csv("@azure_blob_csv_stage/你的文件名.csv")

方式B：手动指定Schema（适合已知字段结构的情况）

如果已经清楚CSV的字段和类型，可以直接定义Schema，这样效率更高：

from snowflake.snowpark.types import StructType, StructField, IntegerType, StringType, FloatType

# 自定义你的CSV Schema
custom_schema = StructType([
    StructField("product_id", IntegerType(), nullable=True),
    StructField("product_name", StringType(), nullable=True),
    StructField("category", StringType(), nullable=True),
    StructField("price", FloatType(), nullable=True)
])

# 读取CSV
product_df = session.read.schema(custom_schema).csv("@azure_blob_csv_stage/你的文件名.csv")

关键说明 #

• 这种方式不会把数据加载到本地内存：所有的读取操作都是在Snowflake的计算仓库中执行的，和Spark的分布式读取逻辑类似，但依托Snowflake的外部存储集成能力。
• 为什么之前直接用session.read.csv('wasbs://...')不行？因为Snowpark没有内置Azure Blob的客户端，必须通过Snowflake的外部阶段来中转访问外部存储，这是和Spark的核心区别。

你可以用product_df.show()来验证读取结果，之后就可以像操作普通Snowpark DataFrame一样进行后续处理了。

备注：内容来源于stack exchange，提问作者Rushyasrunga Kambhampati