使用Python从CSV文件批量生成CREATE TABLE建表SQL语句的实现方法

优先选择pandas实现，比手动转JSON数组再处理的逻辑更简洁，分组和迭代代码都不需要手动实现，后续扩展修改成本也更低。

完整pandas实现代码 #

import pandas as pd

# 1. 读取CSV元数据文件
df = pd.read_csv("metadata.csv")

sql_statements = []
# 2. 按schema、table分组聚合同表的所有字段
for (schema, table), col_group in df.groupby(["SCHEMA", "TABLE"]):
    # 3. 拼接单表的所有字段定义
    col_defs = []
    for _, row in col_group.iterrows():
        col_def = f"{row['COLUMN'].lower()} {row['TYPE'].lower()}({row['LENGTH']})"
        col_defs.append(col_def)
    # 4. 拼接建表语句
    create_sql = f"create table {schema.lower()}.{table.lower()} ({', '.join(col_defs)});"
    sql_statements.append(create_sql)

# 最终合并为单个字符串变量
all_create_sql = "\n".join(sql_statements)

执行后all_create_sql变量存储的内容就和你给出的预期SQL输出完全一致。

无第三方依赖的替代方案 #

如果运行环境无法安装pandas，可以用Python内置的csv模块+字典分组实现，也比手动转JSON数组的逻辑更高效：

import csv
from collections import defaultdict

# 按schema+table为key分组存储字段信息
table_meta = defaultdict(list)

with open("metadata.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        group_key = (row["SCHEMA"], row["TABLE"])
        table_meta[group_key].append(row)

sql_statements = []
for (schema, table), cols in table_meta.items():
    col_defs = [f"{c['COLUMN'].lower()} {c['TYPE'].lower()}({c['LENGTH']})" for c in cols]
    sql_statements.append(f"create table {schema.lower()}.{table.lower()} ({', '.join(col_defs)});")

all_create_sql = "\n".join(sql_statements)

方案对比 #

pandas方案的优势非常明显：

• 无需手动实现分组逻辑，内置groupby方法直接完成表级聚合，避免手动写分组逻辑容易出现的bug
• 自带完善的CSV解析能力，不用自己处理表头映射、特殊字符转义、空值处理等边界场景
• 后续如果要扩展功能（比如过滤指定库的表、修改字段类型映射、增加主键约束等），直接操作DataFrame即可，代码可读性和维护性远高于手动遍历JSON数组

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