如何通过Python将ADLS Blob数据迁移至Snowflake并解决认证报错

问题根因 #

你使用的generate_blob_sas()是单Blob对象级的SAS生成接口，天然不支持容器/目录级的list权限，也无法覆盖目录下的多个文件。Snowflake执行COPY INTO时需要枚举目标路径下的所有文件，缺少list权限就会触发091003错误、返回403认证失败。

实现步骤 #

1. 替换SAS生成逻辑 #

根据权限控制粒度选择对应SAS生成方式：

方案A：容器级SAS（适用容器下所有目录都需要导入的场景）

from azure.storage.blob import generate_container_sas, ContainerSasPermissions
from datetime import datetime, timedelta

# 存储侧配置
account_name = "Azure存储账户名称"
account_key = "Azure存储账户访问密钥"
container_name = "目标容器名称"

# 生成带read、list权限的容器级SAS
sas_token = generate_container_sas(
    account_name=account_name,
    container_name=container_name,
    account_key=account_key,
    permission=ContainerSasPermissions(read=True, list=True),
    expiry=datetime.utcnow() + timedelta(hours=24) # 按需调整有效期
)

# 拼接完整访问路径
adls_path = f"https://{account_name}.blob.core.windows.net/{container_name}?{sas_token}"

方案B：目录级SAS（适用仅需开放指定目录权限的场景，仅ADLS Gen2支持）

from azure.storage.filedatalake import generate_directory_sas, DirectorySasPermissions

sas_token = generate_directory_sas(
    account_name=account_name,
    file_system_name=container_name,
    directory_name="目标导入目录的路径",
    account_key=account_key,
    permission=DirectorySasPermissions(read=True, list=True),
    expiry=datetime.utcnow() + timedelta(hours=24)
)

adls_path = f"https://{account_name}.dfs.core.windows.net/{container_name}/目标导入目录的路径?{sas_token}"

2. 执行Snowflake批量导入 #

import snowflake.connector

# Snowflake连接配置
conn = snowflake.connector.connect(
    user="Snowflake用户名",
    password="Snowflake密码",
    account="Snowflake账户标识",
    warehouse="目标计算仓库",
    database="目标数据库",
    schema="目标Schema"
)

cur = conn.cursor()
try:
    copy_sql = f"""
    COPY INTO 目标表名称
    FROM '{adls_path}/'
    FILE_FORMAT = (
        TYPE = 'CSV'
        SKIP_HEADER = 1 -- 按需调整是否跳过表头
        FIELD_OPTIONALLY_ENCLOSED_BY = '"'
        ENCODING = 'UTF8'
    )
    PATTERN = '.*[.]csv' -- 匹配所有CSV文件，可自定义规则
    ON_ERROR = 'CONTINUE' -- 按需配置错误处理逻辑
    """
    cur.execute(copy_sql)
    # 打印导入结果
    print("导入结果：", cur.fetchall())
finally:
    cur.close()
    conn.close()

注意事项 #

• 生成SAS时统一使用UTC时间计算有效期，避免时区差导致SAS未生效或提前过期
• 若Azure存储账户配置了网络访问限制，需要将Snowflake对应区域的IP段加入存储账户白名单
• 可通过修改PATTERN参数的正则规则，匹配指定命名规则的CSV文件，实现部分文件的批量导入

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