如何用Python将共享文件夹Excel导入Snowflake并导出CSV

一、准备工作 #

• 安装依赖库：pip install pandas snowflake-connector-python openpyxl snowflake-sqlalchemy
• 准备Snowflake核心连接信息：账户名、用户名、密码、目标仓库、数据库、模式、表名

二、读取共享文件夹中的Excel文件 #

（你已实现此步骤，以下是标准示例供参考，注意共享路径的写法：Windows用\\server\share\file.xlsx，Linux/mac用//server/share/file.xlsx）

import pandas as pd

# 共享文件夹路径
excel_path = r"\\shared-server\data_folder\source_data.xlsx"
# 读取指定sheet的数据
df = pd.read_excel(excel_path, sheet_name="Sheet1")

# 可选：预处理数据匹配Snowflake表结构
df = df.dropna(subset=["核心标识列"])  # 清理关键空值
df.columns = [col.upper() for col in df.columns]  # Snowflake默认列名大写，统一格式

三、将数据导入Snowflake #

提供两种常用方案，按需选择：

方案1：用pandas to_sql（适合中小数据量，操作简单） #

from sqlalchemy import create_engine

# 构建Snowflake连接字符串
snowflake_conn_str = (
    "snowflake://{user}:{password}@{account}/{database}/{schema}?warehouse={warehouse}"
).format(
    user="你的用户名",
    password="你的密码",
    account="你的账户名（格式如：xxx.us-west-2）",
    database="目标数据库名",
    schema="目标模式名",
    warehouse="目标仓库名"
)

engine = create_engine(snowflake_conn_str)

# 导入数据：if_exists参数可选replace（覆盖）、append（追加）、fail（表存在则报错）
df.to_sql(
    name="目标表名",
    con=engine,
    if_exists="replace",
    index=False,
    chunksize=10000  # 分块导入，避免内存溢出
)

# 关闭连接释放资源
engine.dispose()

方案2：用Snowflake Connector执行COPY INTO（适合大数据量，效率更高） #

此方案需先将Excel上传至Snowflake临时阶段，再复制到目标表：

import snowflake.connector

# 建立Snowflake连接
conn = snowflake.connector.connect(
    user="你的用户名",
    password="你的密码",
    account="你的账户名",
    warehouse="目标仓库名",
    database="目标数据库名",
    schema="目标模式名"
)
cursor = conn.cursor()

# 1. 创建临时存储阶段
cursor.execute("CREATE OR REPLACE TEMPORARY STAGE excel_temp_stage")

# 2. 上传共享文件夹的Excel到临时阶段（若共享路径无法直接被Snowflake访问，需先复制到本地临时目录）
cursor.execute(f"PUT file://{excel_path} @excel_temp_stage AUTO_COMPRESS=FALSE")

# 3. 将阶段数据复制到目标表（需确保表结构与Excel匹配，可指定列映射）
cursor.execute("""
    COPY INTO 目标表名
    FROM @excel_temp_stage/source_data.xlsx
    FILE_FORMAT = (TYPE = EXCEL SHEET = 'Sheet1')
    ON_ERROR = 'CONTINUE'  # 遇到错误继续执行，可根据需求调整
""")

# 提交操作并关闭连接
conn.commit()
cursor.close()
conn.close()

四、从Snowflake表生成三个CSV文件 #

以下两种常见拆分场景，按需选择：

场景1：按业务条件拆分（如按分类列值拆分） #

import pandas as pd
from sqlalchemy import create_engine

# 复用之前的连接字符串
engine = create_engine(snowflake_conn_str)

# 查询第一类数据并保存
df_category1 = pd.read_sql("SELECT * FROM 目标表名 WHERE 分类列 = '类别A'", con=engine)
df_category1.to_csv("output_categoryA.csv", index=False, encoding="utf-8-sig")

# 查询第二类数据并保存
df_category2 = pd.read_sql("SELECT * FROM 目标表名 WHERE 分类列 = '类别B'", con=engine)
df_category2.to_csv("output_categoryB.csv", index=False, encoding="utf-8-sig")

# 查询第三类数据并保存
df_category3 = pd.read_sql("SELECT * FROM 目标表名 WHERE 分类列 = '类别C'", con=engine)
df_category3.to_csv("output_categoryC.csv", index=False, encoding="utf-8-sig")

engine.dispose()

场景2：按数据量平均拆分（如将全量数据分成3份） #

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine(snowflake_conn_str)

# 查询全量数据
df_full = pd.read_sql("SELECT * FROM 目标表名", con=engine)

# 计算每份数据量
total_rows = len(df_full)
chunk_size = total_rows // 3

# 拆分并保存
df_part1 = df_full.iloc[:chunk_size]
df_part1.to_csv("output_part1.csv", index=False, encoding="utf-8-sig")

df_part2 = df_full.iloc[chunk_size:2*chunk_size]
df_part2.to_csv("output_part2.csv", index=False, encoding="utf-8-sig")

df_part3 = df_full.iloc[2*chunk_size:]
df_part3.to_csv("output_part3.csv", index=False, encoding="utf-8-sig")

engine.dispose()

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