基于 OpenAI API 智能优化数仓表字段类型的实践与思考

# 优化流程示意图
数据采样 → 统计分析 → AI分析 → 类型推荐 → DDL生成 → 验证执行

import pandas as pd
import openai
import json

def sample_data_describe(table_name, sample_size=1000):
    """
    获取表数据样本和统计信息
    """
    # 实际环境中替换为真实的数仓查询代码
    query = f"SELECT * FROM {table_name} TABLESAMPLE SYSTEM (10) WHERE RAND() < 0.1 LIMIT {sample_size}"
    # df = execute_query(query)  # 实际执行查询
    
    # 模拟返回的示例数据
    sample_data = {
        'user_id': ['123', '456', '789', '001', ...],
        'amount': ['100.50', '200.00', '50.25', ...],
        'create_time': ['2023-01-01 10:00:00', '2023-01-02 11:30:00', ...],
        'status': ['1', '0', '1', '1', ...]
    }
    
    df = pd.DataFrame(sample_data)
    return df.describe(include='all').to_dict()

def build_type_optimization_prompt(table_name, column_stats):
    """
    构建类型优化提示词
    """
    prompt = f"""
你是一个资深的数据仓库工程师，请分析以下{table_name}表的字段统计信息，
为每个字段推荐最合适的数据库数据类型。考虑存储效率、查询性能和业务语义。

请按JSON格式返回结果，包含字段名、当前类型、推荐类型、推荐理由和置信度。

字段统计信息：
{json.dumps(column_stats, indent=2)}

要求：
1. 数值类型：根据数值范围和精度选择TINYINT/SMALLINT/INT/BIGINT/DECIMAL
2. 字符串类型：根据长度分布选择VARCHAR/CHAR/TEXT
3. 时间类型：统一时间格式，考虑时区需求
4. 布尔类型：使用最适合的布尔表示方式
5. 考虑字段的语义和业务含义

返回格式示例：
{{
  "recommendations": [
    {{
      "column_name": "example_field",
      "current_type": "VARCHAR(255)",
      "recommended_type": "INT",
      "reason": "该字段只包含数字ID，转换为INT可节省40%存储空间",
      "confidence": 0.9
    }}
  ]
}}
"""
    return prompt

def get_type_recommendations(table_name, column_stats):
    """
    调用OpenAI API获取类型推荐
    """
    openai.api_key = os.getenv("OPENAI_API_KEY")
    
    prompt = build_type_optimization_prompt(table_name, column_stats)
    
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "你是一个专业的数据架构师，擅长数据库性能优化。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1  # 低随机性以保证稳定性
        )
        
        recommendations = json.loads(response.choices[0].message.content)
        return recommendations
        
    except Exception as e:
        print(f"API调用失败: {str(e)}")
        return None

def generate_ddl_statements(table_name, recommendations, original_schema):
    """
    生成DDL修改语句
    """
    ddl_statements = []
    
    for recommendation in recommendations['recommendations']:
        column_name = recommendation['column_name']
        recommended_type = recommendation['recommended_type']
        
        # 生成ALTER TABLE语句
        ddl = f"ALTER TABLE {table_name} MODIFY COLUMN {column_name} {recommended_type};"
        ddl_statements.append({
            'ddl': ddl,
            'reason': recommendation['reason'],
            'confidence': recommendation['confidence']
        })
    
    return ddl_statements

CREATE TABLE orders (
    order_id VARCHAR(255),
    user_id VARCHAR(255),
    amount VARCHAR(100),
    create_time VARCHAR(50),
    status VARCHAR(10)
);

{
  "recommendations": [
    {
      "column_name": "order_id",
      "current_type": "VARCHAR(255)",
      "recommended_type": "BIGINT",
      "reason": "该字段为纯数字订单ID，范围在1-10^9之间，BIGINT足够且节省75%存储",
      "confidence": 0.95
    },
    {
      "column_name": "user_id",
      "current_type": "VARCHAR(255)",
      "recommended_type": "INT",
      "reason": "用户ID为固定8位数字，INT类型足够存储且提高关联性能",
      "confidence": 0.9
    },
    {
      "column_name": "amount",
      "current_type": "VARCHAR(100)",
      "recommended_type": "DECIMAL(10,2)",
      "reason": "金额字段需要精确计算，转换为DECIMAL确保计算准确性",
      "confidence": 0.98
    },
    {
      "column_name": "create_time",
      "current_type": "VARCHAR(50)",
      "recommended_type": "TIMESTAMP",
      "reason": "时间字段应使用专用时间类型，支持时间计算和索引优化",
      "confidence": 0.99
    },
    {
      "column_name": "status",
      "current_type": "VARCHAR(10)",
      "recommended_type": "TINYINT",
      "reason": "状态字段只有0-5六个值，TINYINT足够且节省90%存储",
      "confidence": 0.85
    }
  ]
}

ALTER TABLE orders MODIFY COLUMN order_id BIGINT;
ALTER TABLE orders MODIFY COLUMN user_id INT;
ALTER TABLE orders MODIFY COLUMN amount DECIMAL(10,2);
ALTER TABLE orders MODIFY COLUMN create_time TIMESTAMP;
ALTER TABLE orders MODIFY COLUMN status TINYINT;

# 重要的安全检查
def validate_type_change(original_type, new_type, sample_data):
    """
    验证类型转换的安全性
    """
    # 检查是否会丢失精度
    # 检查是否会截断数据
    # 检查是否会导致数值溢出
    # 返回验证结果和潜在问题
    pass