使用Python为复杂JSON数据自动创建MySQL数据表的方法咨询

实现逻辑 #

核心逻辑分为三步：

1. 先扫描所有待处理的JSON文件，提取全量字段集合，同时根据字段值的类型映射对应的MySQL数据类型
2. 基于汇总的字段集合自动建表，所有字段默认允许为NULL，缺失值插入时自动填充NULL
3. 插入数据时自动补全每个JSON对象缺失的字段，统一字段顺序后批量写入

依赖准备 #

需要提前安装两个依赖库：

• pymysql：用于连接操作MySQL数据库
• pandas：用于JSON数据格式化和批量写入，也可根据需求替换为原生SQL操作
  安装命令：

pip install pymysql pandas

完整实现代码 #

import pymysql
import json
from typing import List, Dict

# 数据库连接配置，修改为自己的实际配置
DB_CONFIG = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "root",
    "password": "your_password",
    "database": "your_database",
    "charset": "utf8mb4"
}

def flatten_json(nested_json: Dict, parent_key: str = '', sep: str = '_') -> Dict:
    """扁平化嵌套JSON，嵌套字段转为 父字段_子字段 的格式"""
    items = []
    for k, v in nested_json.items():
        new_key = parent_key + sep + k if parent_key else k
        if isinstance(v, Dict):
            items.extend(flatten_json(v, new_key, sep=sep).items())
        elif isinstance(v, List) and all(isinstance(i, Dict) for i in v):
            # 对象数组默认转为JSON字符串存储，可根据需求调整处理逻辑
            items.append((new_key, json.dumps(v, ensure_ascii=False)))
        else:
            items.append((new_key, v))
    return dict(items)

def get_all_fields(json_files: List[str]) -> Dict:
    """扫描所有JSON文件，获取全量字段和对应的数据类型"""
    all_fields = {}
    for file_path in json_files:
        with open(file_path, 'r', encoding='utf-8') as f:
            data_list = json.load(f)
            # 单个JSON对象统一转为列表处理
            if isinstance(data_list, Dict):
                data_list = [data_list]
            for item in data_list:
                flat_item = flatten_json(item)
                for k, v in flat_item.items():
                    # 已识别的字段跳过重复判断，也可根据需求做类型兼容逻辑
                    if k in all_fields:
                        continue
                    # 映射MySQL数据类型
                    if isinstance(v, int):
                        all_fields[k] = "INT"
                    elif isinstance(v, float):
                        all_fields[k] = "DOUBLE"
                    elif isinstance(v, bool):
                        all_fields[k] = "TINYINT(1)"
                    else:
                        # 字符串/其他类型默认用TEXT，可根据长度调整为VARCHAR
                        all_fields[k] = "TEXT"
    return all_fields

def create_table(table_name: str, fields: Dict):
    """根据字段集合自动建表"""
    conn = pymysql.connect(**DB_CONFIG)
    cursor = conn.cursor()
    # 默认添加自增主键，可根据需求调整
    create_sql = f"CREATE TABLE IF NOT EXISTS `{table_name}` (`id` INT AUTO_INCREMENT PRIMARY KEY, "
    # 拼接所有字段，默认允许为NULL
    field_sql_list = [f"`{k}` {v} DEFAULT NULL" for k, v in fields.items()]
    create_sql += ", ".join(field_sql_list) + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;"
    cursor.execute(create_sql)
    conn.commit()
    cursor.close()
    conn.close()

def insert_data(table_name: str, json_files: List[str], all_fields: List[str]):
    """批量插入数据，缺失字段自动填NULL"""
    conn = pymysql.connect(**DB_CONFIG)
    all_data = []
    for file_path in json_files:
        with open(file_path, 'r', encoding='utf-8') as f:
            data_list = json.load(f)
            if isinstance(data_list, Dict):
                data_list = [data_list]
            for item in data_list:
                flat_item = flatten_json(item)
                # 按全量字段顺序补全缺失值
                row = [flat_item.get(k, None) for k in all_fields]
                all_data.append(row)
    # 生成批量插入SQL
    placeholders = ", ".join(["%s"] * len(all_fields))
    insert_sql = f"INSERT INTO `{table_name}` ({', '.join([f'`{k}`' for k in all_fields])}) VALUES ({placeholders})"
    cursor = conn.cursor()
    cursor.executemany(insert_sql, all_data)
    conn.commit()
    cursor.close()
    conn.close()

if __name__ == "__main__":
    # 替换为实际的JSON文件路径列表
    JSON_FILES = ["file1.json", "file2.json", "file3.json"]
    TABLE_NAME = "json_data_table"
    # 扫描得到全量字段
    fields_dict = get_all_fields(JSON_FILES)
    all_fields = list(fields_dict.keys())
    # 自动建表
    create_table(TABLE_NAME, fields_dict)
    # 批量插入数据
    insert_data(TABLE_NAME, JSON_FILES, all_fields)

扩展优化说明 #

• 如果需要支持建表后新增字段的场景，可以在插入数据前校验当前表的字段，遇到新字段自动执行ALTER TABLE语句添加字段，默认设为NULL即可
• 如果JSON数据量很大，可以分批读取和写入，避免内存占用过高
• 字段类型映射可以根据业务需求调整，比如长文本可以改为LONGTEXT，固定长度的字符串可以改为VARCHAR类型提升性能

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