如何不使用SQLAlchemy，用Pandas DataFrame在Snowflake建表

可以实现，无需依赖SQLAlchemy或本地文件存储，核心思路是先基于DataFrame Schema自动生成建表SQL，再用Snowflake Connector的write_pandas写入数据（注：write_pandas并非仅支持追加，只要目标表存在即可写入）。

完整实现代码 #

1. 辅助函数：生成Snowflake建表SQL #

根据Pandas DataFrame的数据类型，映射为Snowflake兼容的字段类型，自动生成CREATE TABLE语句：

def generate_create_table_sql(df, table_name, schema=None, database=None):
    # 构建表的全限定名
    full_table_parts = []
    if database:
        full_table_parts.append(database)
    if schema:
        full_table_parts.append(schema)
    full_table_parts.append(table_name)
    full_table_name = ".".join(full_table_parts)
    
    # Pandas dtype到Snowflake数据类型的基础映射
    dtype_mapping = {
        'int64': 'NUMBER(38,0)',
        'float64': 'FLOAT',
        'object': 'STRING',
        'datetime64[ns]': 'TIMESTAMP_NTZ',
        'bool': 'BOOLEAN',
        'datetime64[ns, UTC]': 'TIMESTAMP_LTZ'
    }
    
    # 生成字段定义列表
    column_defs = []
    for col_name, dtype in df.dtypes.items():
        sf_type = dtype_mapping.get(str(dtype), 'STRING')  # 未知类型默认用STRING兜底
        column_defs.append(f'"{col_name}" {sf_type}')
    
    return f"CREATE OR REPLACE TABLE {full_table_name} ({', '.join(column_defs)})"

2. 主逻辑：连接、数据处理、建表与写入 #

import configparser
import pandas as pd
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas

# 读取配置文件
config = configparser.ConfigParser()
config.read('db.ini')

# 提取连接参数
sn_user = config['database_name']['user']
sn_password = config['database_name']['pass']
sn_account = config['database_name']['acc']
sn_warehouse = config['database_name']['wh']
sn_database = config['database_name']['db']
sn_schema = config['database_name']['db_schema']

# 建立Snowflake连接
ctx = snowflake.connector.connect(
    user=sn_user,
    password=sn_password,
    account=sn_account,
    warehouse=sn_warehouse,
    database=sn_database,
    schema=sn_schema
)
cs = ctx.cursor()

# 数据提取与清洗（保留原有逻辑）
query_extract = '''
    select table1.field1, table1.field2, table1.field3,
           table2.field2, table2.field5, table3.field1
    from database.schema.table1
    left join database.schema.table2 on table1.field3 = table2.field1
    left join database.schema.table3 on table1.field5 = table3.field1
'''
try:
    cs.execute(query_extract)
    df = cs.fetch_pandas_all()
    # 替换为你的实际数据清洗逻辑
    df_final = df.drop_duplicates().reset_index(drop=True)
except Exception as e:
    print(f"数据处理失败: {str(e)}")
    ctx.close()
    exit()

# 目标表名称
TARGET_TABLE = "PROCESSED_DATA_TABLE"

# 步骤1：生成并执行建表SQL
create_table_sql = generate_create_table_sql(df_final, TARGET_TABLE, sn_schema, sn_database)
cs.execute(create_table_sql)

# 步骤2：用write_pandas写入数据（无需本地文件）
success, chunk_count, row_count, _ = write_pandas(
    conn=ctx,
    df=df_final,
    table_name=TARGET_TABLE,
    database=sn_database,
    schema=sn_schema,
    auto_create_table=False  # 已手动建表，关闭自动创建
)

print(f"数据写入结果: 成功={success}, 写入行数={row_count}")

# 关闭连接
ctx.close()

关键说明 #

• write_pandas的真实能力：它会自动将DataFrame数据上传到Snowflake临时阶段，再执行COPY INTO语句，支持写入新创建的空表，并非仅能追加。
• 数据类型映射可扩展：如果需要更精细的类型控制（比如指定NUMBER的精度、DATE类型），可以修改dtype_mapping字典。
• 全程无本地文件：数据在内存中完成流转，无需落地存储。

提问优化建议 #

1. 明确约束的边界：补充说明拒绝SQLAlchemy的具体原因（比如依赖冲突、性能需求、环境限制），帮助回答者精准匹配方案。
2. 简化示例代码：将冗长的查询语句简化为占位符（比如SELECT * FROM sample_table），减少非核心信息干扰。
3. 补充预期细节：说明是否需要保留DataFrame的字段大小写、是否需要自定义主键/约束，让方案更贴合实际需求。

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