第四章：结构化数据导入——从CSV、Excel到数据库
数据准备好了，接下来就是怎么把它们喂进知识图谱管道。我看到很多新手拿到Excel就往里塞，结果跑出来的实体全是乱码，或者关系丢了一半。嗯，这章我们就来聊聊结构化数据导入，说白了就是怎么让表格里的每行每列，都变成图谱里的节点和边。
4.1 数据源摸底——先搞清楚你面对的是什么
动手写代码前，我建议你先做个简单摸底。不同数据源，处理策略完全不同。我自己一般先看三个维度：格式是否规整、有没有表头、编码是什么。
下面这张图，可以帮助你快速定位当前数据源适合用什么方式接入：
为什么要先做摸底？因为结构化数据来源不同，坑也不一样。
| 数据源类型 | 常见问题 | 我的处理习惯 |
|---|---|---|
| CSV | 编码混乱、分隔符不统一、有空行 | 先看一眼文件头，用命令行快速校验 |
| Excel | 合并单元格、公式残留、多Sheet | 先转CSV，除非有格式要求 |
| JSON | 嵌套太深、结构不一致 | 用jq展平后再处理 |
| 关系数据库 | 外键约束、大表分页、权限限制 | 直接读视图，不碰原表 |
个人习惯：拿到Excel先看右下角"数据"选项卡里的"查询和连接"——如果发现有外部链接，一定要问清楚数据是哪里来的。我在项目中遇到过，一个Excel里嵌了37个Power Query外部链接，导出时全部断链，数据直接少了一半。
4.2 CSV与TXT——最朴素也最容易翻车
CSV是知识图谱导入的"基础设施"。但你想想看，人人都在用CSV，出错的概率反而最大。
核心步骤只有两步：读文件，按实体-关系拆列。但细节决定了成败。
import csv
import json
def parse_csv_to_triples(file_path, delimiter=','):
"""
将CSV解析为三元组列表
返回格式：[('实体A', '关系', '实体B'), ...]
"""
triples = []
with open(file_path, 'r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f, delimiter=delimiter)
if not reader.fieldnames:
raise ValueError("没有检测到表头，请确认文件格式")
for row in reader:
# 跳过完全空的行
if not any(row.values()):
continue
# 假设列: subject, predicate, object
subj = row.get('subject', '').strip()
pred = row.get('predicate', '').strip()
obj = row.get('object', '').strip()
if subj and pred and obj:
triples.append((subj, pred, obj))
return triples这里有个关键点：我建议每次都显式指定 encoding='utf-8-sig'。为什么？因为Windows Excel导出的CSV默认是带BOM的，不带-sig你会读到\uFEFF乱码。我曾经被这个问题坑了整整半天。
4.3 Excel——比想象的难伺候
Excel相比CSV多了一层表结构。我个人更推荐用openpyxl而不是xlrd——后者已经不更新了。处理Excel的要点在于：把每个Sheet当成一个独立的表来看。
from openpyxl import load_workbook
def excel_to_entity_table(file_path, sheet_name):
wb = load_workbook(file_path, data_only=True)
if sheet_name not in wb.sheetnames:
raise KeyError(f"Sheet '{sheet_name}' 不存在，现有Sheet: {wb.sheetnames}")
ws = wb[sheet_name]
rows = []
for row in ws.iter_rows(min_row=2, values_only=True):
# 去掉整行都是None的情况
if any(cell is not None for cell in row):
rows.append(tuple(cell for cell in row))
return rows注意：data_only=True这个参数绝对不能省略。如果不加，一旦Excel里有VLOOKUP或者SUMIF这类公式，你拿到的会是公式字符串而不是计算结果。我就因为这个，把一个包含8000条计算公式的预算表直接当原始数据塞进了图谱，结果所有数值都带上了"="前缀。
4.4 关系数据库——数据量一上来，就要讲策略
数据库导入和文件导入不同。文件可以全量加载，但数据库不行——几千万行直接select *是会被运维约谈的。
我的做法是：分页查询 + 批次写入。每页大小一般设5000到10000，太多容易OOM，太少交互太频繁。
import pymysql
import pandas as pd
def fetch_data_in_batches(conn, table_name, batch_size=5000):
offset = 0
while True:
query = f"SELECT * FROM {table_name} LIMIT {batch_size} OFFSET {offset}"
df = pd.read_sql(query, conn)
if df.empty:
break
yield df.to_dict(orient='records')
offset += batch_size如果表里数据量特别大（比如上亿），我建议改用游标式查询。Python的pymysql.cursors.SSCursor可以做到一次只加载一批数据到内存，而不是把整个结果集拖下来。你在写生产管道时一定要用这个，否则内存会像漏了一样涨。
4.5 数据映射——把列变成图谱的"语言"
结构化数据读进来之后，需要做一次"翻译"。表里的每一列，要对应到知识图谱里的什么？
一般来说，我遵循这个映射规则：
- ID列 → 实体唯一标识（URI或UUID）
- 名称列 → 实体标签属性（如rdfs:label）
- 外键列 → 实体间关系（如:works_in）
- 属性列 → 实体属性（如:birthDate）
举个例子，假如有一个员工表：
员工表 (employee)
| employee_id | name | department_id | hire_date |
|-------------|------|---------------|-----------|
| E001 | 张三 | D01 | 2023-01-01 |
映射规则：
- employee_id → :Employee 实体的唯一标识
- name → rdfs:label 属性
- department_id → :works_in 关系（指向:Department实体）
- hire_date → :hireDate 属性映射这一步最容易出问题的是"数据对齐"。比如department_id在员工表里是字符串"D01"，但在部门表里是整数"1"。不统一的话，关系永远连不上。我建议在做映射前，先统计一下所有外键列的值范围和类型，确保两端对齐。
4.6 编码与类型——两大最常见的"隐形杀手"
说了这么多，其实最烦人的往往是编码和类型。这两个问题不解决，数据注入管道后会一路报错到底。
| 问题类别 | 表现 | 解决方法 |
|---|---|---|
| 编码不一致 | 文本显示为乱码或"?" | 统一使用utf-8编码，CSV写入时指定utf-8-sig |
| 数字类型混用 | "1,234"和"1234"并存 | 去除逗号、统一为浮点数或整数 |
| 日期格式不统一 | "2023/01/01"和"01-01-2023"混排 | 使用datetime模块统一解析，最好转ISO 8601格式 |
| 空值处理 | null、"N/A"、空字符串三中情况都有 | 建一个"空值名单"，先把所有值标准化为None |
还有一个坑：不同数据库对NULL的表达不同。MySQL里是NULL，Oracle里就是空字符串''，SQLite又可能是None。我以前写过一个管道，本地SQLite用着好好的，换到线上MySQL直接炸了——就因为空值判断逻辑没做适配。后来我写了个统一空值处理函数，先把所有已知空值形式转成Python的None，再往前走。
4.7 小批量验证——快速反馈，别等跑完才发现
数据导入流程搭好之后，我从来不会一次性全量跑。我的做法是：先拿100条数据跑一次，看一眼输出结果是不是预期的。如果没问题，再跑1000条，最后才是全量。
验证的时候，重点关注三点：
- 实体ID是否唯一？ 重复的ID会导致图谱节点合并失败
- 关系两端是否存在？ 指向不存在的实体的关系会被丢弃
- 属性值类型是否正确？ 日期串当字符串存没问题，但如果你后续要做时间推理，就别偷懒
好了，结构化数据导入就聊到这里。你可以先找一份实际的CSV试着手动导入一次，踩过的坑会记得更牢。