AI 辅助数据探索：未知数据集的结构化描述自动生成-育师

AI 辅助数据探索：未知数据集的结构化描述自动生成

一、拿到一张"三无"表——没文档、没注释、没人懂

干数据分析这行最怕的场景是什么？不是数据量大，不是计算复杂，而是你接了一个完全陌生的数据集，没有任何说明文档。

上个月接手了一个离职同事留下的数据管道，核心表有 60 多个字段，字段名全是f001、f002、attr_12这种"密码学命名"。注释留白，数据字典找不到，连口头交接都没有——同事已经去新公司了。

这种"三无数据集"在数据行业真的太常见了。快速看懂一张陌生表，传统做法是手工探查：SELECT COUNT(*)、SELECT DISTINCT、DESCRIBE TABLE……先花半小时看分布，再花半小时翻业务文档（如果有的话），再花半小时猜字段含义。效率低不说，还容易漏掉隐藏在数据里的异常。

而这个场景恰好是 AI 非常擅长的——它不是替代人工判断，而是帮人工把探查过程加速 10 倍。

二、AI 探索数据集的四层分析框架

AI 阅读一张数据表，和我们人肉读表其实是一个逻辑，但它可以并行、不遗漏、自动生成结构化输出：

flowchart TD A[未知数据集输入] --> B[第一层: Schema 解析] A --> C[第二层: 统计分析] A --> D[第三层: 关联推断] A --> E[第四层: 语义生成] B --> B1[字段名 / 类型 / 注释] B1 --> B2[字段分类: 维度/指标/时间/ID] C --> C1[缺失率 / 唯一值 / 分布形态] C1 --> C2[数据质量标记] D --> D1[字段间相关性矩阵] D1 --> D2[外键候选 / 层级关系 / 1对N] E --> E1[LLM 语义理解] E1 --> E2[自动生成数据字典] B2 --> F[汇总生成结构化数据描述文档] C2 --> F D2 --> F E2 --> F

第一层：Schema 解析 —— 看"骨架"

这是最浅的一层，但也是最容易被忽视的一层。拿到一个 DataFrame，先把它"拆骨"：

字段类型分布：多少 int、多少 string、多少 datetime？这决定了后续的分析策略
字段命名模式：有没有明显的前缀（dim_、fct_）、后缀（_id、_cnt）暗示？
是否有定义约束：Nullable 与否、默认值、注释（如果有的话）

第二层：统计分析 —— 看"血肉"

Schema 是骨架，统计分析是血肉。每个字段过一遍五维画像：

维度	含义	决策价值
缺失率	空值占比	高缺失 = 数据采集盲区
唯一率	去重值 / 总行数	接近 1.0 = 可能是 ID/Key
偏度	分布倾斜程度	严重偏斜 = 可能有极端值
异常率	超出合理范围的比例	数据质量问题高发区
基尼系数	分布均匀度	接近 0 = 均匀（如流水号），接近 1 = 集中

第三层：关联推断 —— 看"关系"

孤立的字段没什么意义，字段之间的关联才构成业务语义。AI 可以自动检测：

强线性相关（Pearson r > 0.9）→ 可能是冗余字段或派生字段
功能依赖（A → B，A 的每个值对应唯一的 B）→ 可能是层级关系，如city_id → province_id
基数比例（1:1、1:N、M:N）→ 判断是否适合做 Join Key

第四层：语义生成 —— 看"灵魂"

前三层是统计驱动，第四层是语义驱动。把前三层的结果喂给 LLM，让它结合字段命名规则和数据分布特征，推断字段的实际业务含义并生成自然语言描述。

三、从 Schema 到语义：让 AI 读出字段的业务含义

这一步需要一点 prompt 工程技巧。单纯的 prompt "帮我描述这张表"效果很差——AI 容易瞎编。正确做法是先喂结构化统计结果，再让 AI 做"填空"：

你是一个数据分析专家。以下是一张电商订单表的结构化探查报告。 请根据统计特征推断每个字段的业务含义，并生成中文数据字典。 字段名称: f01 字段类型: INT 缺失率: 0% 唯一值率: 100% 样例值: 1000234, 1000235, 1000236... 分布特征: 递增、无间断 推断: f01 大概率是自增主键（订单 ID） --- 字段名称: f15 字段类型: DECIMAL(10,2) 缺失率: 2.3% 唯一值率: 85% 样例值: 29.90, 199.00, 59.90, 1299.00... 分布特征: 右偏分布，集中在 20~200 区间，有少量高值 > 1000 推断: f15 很可能是订单金额，右偏分布符合电商消费特征 ---

四、实战：自动化数据探查 Pipeline

完整的数据探索自动化引擎实现：

""" 未知数据集自动探查引擎 输入: 一张不知道干什么用的 DataFrame 输出: 一份结构化的数据字典 + 数据质量报告 """ import pandas as pd import numpy as np from scipy import stats from typing import Dict, List, Tuple import warnings warnings.filterwarnings('ignore') class DataExplorer: """ 自动化数据探索器 使用方式: explorer = DataExplorer(df, table_name='order_info') report = explorer.full_report() """ def __init__(self, df: pd.DataFrame, table_name: str = 'unknown_table'): self.df = df self.table_name = table_name self.n_rows, self.n_cols = df.shape def schema_analysis(self) -> pd.DataFrame: """第一层：Schema 解析 —— 字段基础画像""" records = [] for col in self.df.columns: dtype = self.df[col].dtype # 判断字段大类 if pd.api.types.is_integer_dtype(dtype): field_category = '整数型' elif pd.api.types.is_float_dtype(dtype): field_category = '浮点型' elif pd.api.types.is_datetime64_any_dtype(dtype): field_category = '时间型' else: field_category = '文本型' records.append({ '字段名': col, '数据类型': str(dtype), '字段分类': field_category, '可为空': '是' if self.df[col].isna().any() else '否', }) return pd.DataFrame(records) def statistical_analysis(self) -> pd.DataFrame: """第二层：统计分析 —— 五维画像""" records = [] for col in self.df.columns: series = self.df[col] # 缺失率 missing_rate = series.isna().mean() # 唯一值率 unique_rate = series.nunique() / (len(series) - series.isna().sum()) \ if len(series) > series.isna().sum() else 1.0 # 数值型字段的统计量 if pd.api.types.is_numeric_dtype(series): # 偏度：衡量数据分布的对称性 # 偏度 = 0 → 正态分布 # 偏度 > 0 → 右偏（大头在左边，长尾在右边） # 偏度 < 0 → 左偏（大头在右边，长尾在左边） cleaned = series.dropna() if len(cleaned) > 3: skewness = cleaned.skew() # 四分位数 q1, median, q3 = cleaned.quantile([0.25, 0.5, 0.75]) # 异常值比例（IQR 方法：超过 Q3 + 1.5*IQR 或低于 Q1 - 1.5*IQR） iqr = q3 - q1 outlier_rate = ((cleaned < q1 - 1.5 * iqr) | (cleaned > q3 + 1.5 * iqr)).mean() else: skewness, q1, median, q3, outlier_rate = np.nan, np.nan, np.nan, np.nan, np.nan else: skewness, q1, median, q3, outlier_rate = np.nan, np.nan, np.nan, np.nan, np.nan records.append({ '字段名': col, '缺失率': f'{missing_rate:.1%}', '唯一值率': f'{unique_rate:.1%}', '中位数': round(median, 2) if not np.isnan(median) else '-', '偏度': round(skewness, 2) if not np.isnan(skewness) else '-', '异常值率': f'{outlier_rate:.1%}' if not np.isnan(outlier_rate) else '-', }) return pd.DataFrame(records) def correlation_analysis(self, threshold: float = 0.8) -> List[Dict]: """第三层：关联推断 —— 字段间相关性和依赖关系""" numeric_cols = self.df.select_dtypes(include=[np.number]).columns if len(numeric_cols) < 2: return [] # 计算皮尔逊相关系数矩阵 corr_matrix = self.df[numeric_cols].corr() strong_pairs = [] for i, col1 in enumerate(numeric_cols): for col2 in numeric_cols[i+1:]: r = corr_matrix.loc[col1, col2] if abs(r) >= threshold: strong_pairs.append({ '字段1': col1, '字段2': col2, '相关系数': round(r, 3), '关系类型': '正相关' if r > 0 else '负相关' }) return strong_pairs def detect_potential_keys(self) -> Dict: """智能推断：识别可能的 ID 字段、外键、维度字段""" findings = { 'primary_key_candidates': [], # 主键候选 'foreign_key_candidates': [], # 外键候选 'dimension_fields': [], # 维度字段候选 'metric_fields': [], # 指标字段候选 'timestamp_fields': [], # 时间字段候选 } for col in self.df.columns: series = self.df[col] unique_rate = series.nunique() / (len(series) - series.isna().sum()) \ if len(series) > series.isna().sum() else 0 # 主键候选：唯一率 100%，无缺失 if unique_rate == 1.0 and series.isna().sum() == 0: findings['primary_key_candidates'].append(col) # 外键候选：唯一率不太高也不太低（5%~60%），且有 _id 后缀 col_lower = col.lower() if 0.05 <= unique_rate <= 0.6 and ('_id' in col_lower or col_lower.endswith('id')): findings['foreign_key_candidates'].append(col) # 时间字段候选 if pd.api.types.is_datetime64_any_dtype(series): findings['timestamp_fields'].append(col) elif 'date' in col_lower or 'time' in col_lower or 'dt' in col_lower: findings['timestamp_fields'].append(col) # 指标字段：数值型 + 唯一率不太高 + 不是 ID if pd.api.types.is_numeric_dtype(series) and unique_rate > 0.1: if col not in findings['primary_key_candidates']: findings['metric_fields'].append(col) # 维度字段候选：文本型 + 唯一值不多（< 总行数的 30%） if series.dtype == 'object' and 0 < unique_rate < 0.3: findings['dimension_fields'].append(col) return findings def generate_data_dict(self) -> pd.DataFrame: """ 汇总生成结构化数据字典 这就是可以直接给业务同学看的"表说明书" """ schema = self.schema_analysis() stats = self.statistical_analysis() keys = self.detect_potential_keys() corr = self.correlation_analysis() # 合并 Schema 和统计结果 data_dict = schema.merge(stats, on='字段名', how='left') # 追加推断的角色标签 data_dict['角色推断'] = '' for idx, row in data_dict.iterrows(): col = row['字段名'] roles = [] if col in keys['primary_key_candidates']: roles.append('主键') if col in keys['foreign_key_candidates']: roles.append('外键') if col in keys['dimension_fields']: roles.append('维度') if col in keys['metric_fields']: roles.append('指标') if col in keys['timestamp_fields']: roles.append('时间') if not roles: roles.append('未知') data_dict.at[idx, '角色推断'] = '、'.join(roles) # 追加关联关系 data_dict['关联关系'] = '' for corr_item in corr: for col in [corr_item['字段1'], corr_item['字段2']]: mask = data_dict['字段名'] == col existing = data_dict.loc[mask, '关联关系'].values[0] if mask.any() else '' other = corr_item['字段1'] if col == corr_item['字段2'] else corr_item['字段2'] rel = f"与{other}{corr_item['关系类型']}(r={corr_item['相关系数']})" data_dict.loc[mask, '关联关系'] = (existing + '；' + rel if existing else rel) return data_dict def full_report(self) -> Dict: """生成完整的探查报告""" report = { '表名': self.table_name, '行数': self.n_rows, '列数': self.n_cols, '数据字典': self.generate_data_dict(), '新增_强关联字段对': self.correlation_analysis(), '新增_推断的字段角色': self.detect_potential_keys(), } return report # ============ 使用示例 ============ # 模拟一张未知的数据表 np.random.seed(42) unknown_df = pd.DataFrame({ 'id': range(1, 1001), # 自增 ID 'order_id': ['ORD' + str(i).zfill(6) for i in range(1, 1001)], 'user_id': np.random.choice(range(1, 201), 1000), # 200 个用户的订单 'product_cate': np.random.choice(['食品', '日用品', '电子产品'], 1000, p=[0.4, 0.35, 0.25]), 'amount': np.random.lognormal(mean=4.5, sigma=0.6, size=1000), # 对数正态分布（电商常见） 'order_date': pd.date_range('2026-01-01', periods=1000, freq='H'), 'status': np.random.choice(['completed', 'pending', 'cancelled'], 1000, p=[0.85, 0.10, 0.05]), 'discount_rate': np.random.beta(2, 8, 1000), # 折扣率（Beta分布，集中在低值） }) # 自动探查 explorer = DataExplorer(unknown_df, table_name='user_orders') report = explorer.full_report() # 输出数据字典（核心产出） print(f"📊 表名: {report['表名']}") print(f"📏 规模: {report['行数']} 行 × {report['列数']} 列") print("\n===== 自动生成的字段推断 =====") for role_type, fields in report['新增_推断的字段角色'].items(): if fields: print(f" {role_type}: {fields}") print("\n===== 数据字典 =====") print(report['数据字典'].to_string(index=False)) if report['新增_强关联字段对']: print("\n===== 强关联字段对 (>0.8) =====") for pair in report['新增_强关联字段对']: print(f" {pair['字段1']} ↔ {pair['字段2']}: {pair['关系类型']} ({pair['相关系数']})")

运行这份代码，你会得到一份自动生成的数据字典，清楚标注了哪些列可能是主键、哪些是外键、哪些是维度和指标字段。更重要的是，它没有依赖任何业务文档——完全从数据本身推断出结构信息。