在Pandas数据框中提取SQL Server表的CREATE TABLE语句（DDL）以迁移至Snowflake

我明白你要做的事——把SQL Server里的表结构（包括完整的CREATE TABLE语句，含列、数据类型、主键外键这些细节）提取到Pandas DataFrame里，方便后续迁移到Snowflake。这事儿我帮你拆解成几个可行的步骤：

步骤1：建立SQL Server数据库连接 #

首先得用Python连接到你的SQL Server，推荐用pyodbc或者sqlalchemy，这里用pyodbc举个例子：

import pyodbc
import pandas as pd

# 替换成你的SQL Server连接信息
conn_str = (
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=你的服务器名;"
    "DATABASE=目标数据库名;"
    "UID=用户名;"
    "PWD=密码;"
)

# 建立连接
conn = pyodbc.connect(conn_str)

步骤2：编写SQL查询生成完整DDL #

SQL Server的系统视图里存了所有表的元数据，我们可以通过拼接这些元数据来生成包含列定义、主键、外键的完整CREATE TABLE语句。

主表结构+主键的查询 #

这个查询会生成基础的CREATE TABLE语句，包含列名、数据类型、非空约束和主键：

SELECT 
    DB_NAME() AS db_name,
    s.name AS schema_name,
    t.name AS table_name,
    'CREATE TABLE ' + QUOTENAME(s.name) + '.' + QUOTENAME(t.name) + '(' + 
    -- 拼接列定义
    STRING_AGG(
        QUOTENAME(c.name) + ' ' + 
        CASE 
            WHEN tp.name IN ('varchar', 'nvarchar', 'char', 'nchar') 
                THEN tp.name + '(' + CASE WHEN c.max_length = -1 THEN 'MAX' ELSE CAST(c.max_length AS VARCHAR) END + ')'
            WHEN tp.name IN ('decimal', 'numeric') 
                THEN tp.name + '(' + CAST(c.precision AS VARCHAR) + ',' + CAST(c.scale AS VARCHAR) + ')'
            ELSE tp.name
        END + 
        CASE WHEN c.is_nullable = 0 THEN ' NOT NULL' ELSE ' NULL' END,
        ', '
    ) + 
    -- 拼接主键约束
    CASE 
        WHEN kc.name IS NOT NULL 
            THEN ', CONSTRAINT ' + QUOTENAME(kc.name) + ' PRIMARY KEY (' + STRING_AGG(QUOTENAME(cc.name), ', ') + ')' 
        ELSE '' 
    END + 
    ')' AS create_table_statement
FROM 
    sys.tables t
JOIN 
    sys.schemas s ON t.schema_id = s.schema_id
JOIN 
    sys.columns c ON t.object_id = c.object_id
JOIN 
    sys.types tp ON c.system_type_id = tp.system_type_id AND c.user_type_id = tp.user_type_id
LEFT JOIN 
    sys.key_constraints kc ON t.object_id = kc.parent_object_id AND kc.type = 'PK'
LEFT JOIN 
    sys.index_columns ic ON kc.parent_object_id = ic.object_id AND kc.unique_index_id = ic.index_id
LEFT JOIN 
    sys.columns cc ON ic.object_id = cc.object_id AND ic.column_id = cc.column_id
-- 如果你只需要特定表，就保留WHERE条件；要全部表就删掉
WHERE 
    t.name IN ('表1', '表2')
GROUP BY 
    s.name, t.name, kc.name
ORDER BY 
    s.name, t.name

补充外键约束（可选） #

如果需要把外键也包含进去，可以用下面的查询生成ALTER TABLE语句，之后你可以把它合并到主DDL里，或者作为DataFrame的单独列：

SELECT 
    DB_NAME() AS db_name,
    s.name AS schema_name,
    t.name AS table_name,
    'ALTER TABLE ' + QUOTENAME(s.name) + '.' + QUOTENAME(t.name) + ' ADD CONSTRAINT ' + QUOTENAME(fk.name) + ' FOREIGN KEY (' + STRING_AGG(QUOTENAME(c.name), ', ') + ') REFERENCES ' + QUOTENAME(ref_s.name) + '.' + QUOTENAME(ref_t.name) + '(' + STRING_AGG(QUOTENAME(ref_c.name), ', ') + ')' AS foreign_key_statement
FROM 
    sys.foreign_keys fk
JOIN 
    sys.tables t ON fk.parent_object_id = t.object_id
JOIN 
    sys.schemas s ON t.schema_id = s.schema_id
JOIN 
    sys.tables ref_t ON fk.referenced_object_id = ref_t.object_id
JOIN 
    sys.schemas ref_s ON ref_t.schema_id = ref_s.schema_id
JOIN 
    sys.foreign_key_columns fkc ON fk.object_id = fkc.constraint_object_id
JOIN 
    sys.columns c ON fkc.parent_object_id = c.object_id AND fkc.parent_column_id = c.column_id
JOIN 
    sys.columns ref_c ON fkc.referenced_object_id = ref_c.object_id AND fkc.referenced_column_id = ref_c.column_id
GROUP BY 
    s.name, t.name, fk.name, ref_s.name, ref_t.name

步骤3：加载结果到Pandas DataFrame #

把上面的SQL查询放到Python里执行，结果直接转成DataFrame：

# 主表结构查询（替换成你实际的SQL语句）
main_sql = """
-- 这里放上面的主表结构+主键的SQL
"""

# 读取数据到DataFrame
df = pd.read_sql(main_sql, conn)

# 如果你需要外键，可以单独读取合并
# fk_sql = """-- 外键查询SQL"""
# df_fk = pd.read_sql(fk_sql, conn)
# df = pd.merge(df, df_fk, on=['db_name', 'schema_name', 'table_name'], how='left')

# 关闭连接
conn.close()

# 查看最终的DataFrame
print(df)

额外提示：适配Snowflake数据类型 #

因为SQL Server和Snowflake的数据类型不完全一致，迁移前记得调整DDL里的类型映射，比如：

• SQL Server VARCHAR(MAX) → Snowflake VARCHAR
• SQL Server DATETIME → Snowflake TIMESTAMP_NTZ
• SQL Server INT → Snowflake INT
• SQL Server DECIMAL(p,s) → Snowflake DECIMAL(p,s)

你可以写个简单的字符串替换函数来批量处理这些转换。

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