📚 课程目录第3章数据源分析与采集策略

3. 数据源分析与采集策略

数据源分析，说白了就是搞清楚你的知识图谱原材料从哪来、长什么样、能不能用。我早期做过一个医疗知识图谱项目，团队上来直接拿了一堆PDF病历就开始抽实体，结果抽出来的数据一半是乱码——因为PDF里既有扫描图片又有表格，字符串编码还不统一。那之后我才真正明白，不把数据源摸透，后面的管道就是空中楼阁。

这一章我们聚焦两个核心问题：数据源到底有哪些类型？针对每种类型该怎么设计采集策略？ 嗯，这里我要先泼一盆冷水：没有万能采集方案，每种数据源都有自己的脾气。

3.1 数据源类型与特点

我把常见的数据源粗分为三大类。你在项目中可能遇到更细的划分，但这张表基本够用了。

类型	典型代表	结构程度	采集难度
结构化	关系数据库（MySQL、PostgreSQL）、CSV、Excel	高（表、行、列、主外键）	低
半结构化	JSON、XML、YAML、日志文件、NoSQL数据库（MongoDB、Elasticsearch）	中（自带标签/键值对，但无固定模式）	中等
非结构化	PDF、Word、HTML网页、Markdown、纯文本、图片中的文字	低（无预定义结构，需要NLP或OCR）	高

千万别以为结构化数据就绝对干净。我在项目中遇到过关系表里大量外键指向不存在的记录，这种“脏结构”比非结构化数据更坑人，你得自己写脚本去清洗。

3.2 采集策略设计

选策略前先问自己两个问题：全量还是增量？实时还是批量？

全量采集：适合一次性建图或数据源变更频率极低。但如果你每天跑全量，数据量大时效率就是灾难。
增量采集：靠时间戳、版本号或CDC（Change Data Capture）来抓取变动部分。我建议对业务系统优先用增量，用updated_at字段做游标就能搞定80%的场景。
批量vs实时：知识图谱管道通常都是批处理（比如每小时或每天调度）。除非你的业务需要秒级更新，否则别给自己找麻烦。

一个实用的设计原则：采集层和清洗层要分开。采集只负责“拿到数据”，不改变格式；清洗在下游做。这样即使数据源头换了，你只需要换采集接口，管道其他部分不受影响。

下面我给出一个简单的API采集示例，抓取开源项目的Issue列表（JSON格式）：

import requests
import json
import time

API_URL = "https://api.github.com/repos/apache/incubator-kyuubi/issues"
HEADERS = {"Accept": "application/vnd.github.v3+json"}
PARAMS = {"state": "all", "per_page": 100, "page": 1}

all_issues = []
while True:
    resp = requests.get(API_URL, headers=HEADERS, params=PARAMS)
    if resp.status_code != 200:
        print(f"请求失败: {resp.status_code}")
        break
    data = resp.json()
    if not data:
        break
    all_issues.extend(data)
    PARAMS["page"] += 1
    time.sleep(0.5)  # 避免触发限流

# 保存为JSON行文件
with open("issues.jsonl", "w", encoding="utf-8") as f:
    for issue in all_issues:
        f.write(json.dumps(issue, ensure_ascii=False) + "\n")

如果你面对是关系数据库，用SQL做增量采集更直接：

-- 假设有个 update_time 字段，上次最大时间戳记在 last_sync_time
SELECT * FROM orders
WHERE update_time > '2025-03-01 00:00:00'
  AND update_time <= '2025-03-02 00:00:00';

小心时区陷阱。我曾经因为数据库服务器和采集服务器时区不一致，导致增量丢失了一整天的数据。后来所有时间戳一律转为UTC存储，接口统一传UTC时间戳字符串。

3.3 质量评估与清洗

数据采集进来后，别急着送进知识图谱管道。先做一轮快速质量检查：

完整性：关键字段是否有空值？比如实体名称、关系类型不能为空。
一致性：同一实体的属性在不同文档中是否矛盾？例如“张三”在A表是男性，在B表是女性。
时效性：数据是否过期？过时关系会污染图谱。

我常用的一个“五秒检查法”：用pandas读入数据，打印df.info()和df.describe(include='all')，一眼就能看出空值率、重复行、离谱的异常值。如果你发现某个字段90%都是缺失，那这个字段在知识图谱里基本就是废的，直接丢掉比费力填充好。

3.4 自动化采集管道

手动采集不可能持续。我建议把采集任务封装成独立的服务或脚本，用调度工具（比如Apache Airflow或简单的crontab）定期触发。下面是一个简化版的管道配置示例，用YAML定义采集任务：

sources:
  - name: mysql_orders
    type: jdbc
    connection: "jdbc:mysql://host:3306/orders"
    query: "SELECT id, user_id, product_id, created_at FROM orders WHERE created_at > :last_sync"
    incremental_column: "created_at"
    output_format: "jsonl"
    output_path: "/data/landing/orders/"

  - name: github_issues
    type: api
    endpoint: "https://api.github.com/repos/apache/incubator-kyuubi/issues"
    params:
      state: "all"
      per_page: 100
    pagination: "page"
    output_format: "jsonl"
    output_path: "/data/landing/issues/"

每个采集任务执行完后，把原始数据打上时间戳，放到/data/landing/目录的对应子目录里。下游清洗模块用watchdog或者定时扫描新文件来触发下一步。

采集中间数据不要放在临时目录。我曾为了省事直接用内存队列传递，结果采集进程挂了，数据全部丢失，还得重新拉一次。落盘到分布式文件系统（比如HDFS或S3）是最保险的做法，哪怕清洗失败也能重新跑。

为了让整个思路更直观，我画了一张流程图，把数据源分析到采集的核心步骤串起来：

画里的③质量清洗与落地，实际上包含了我在3.3节讲的那些检查步骤。别小看这个节点，我见过好几个团队跳过它，结果图谱里的实体数量是上去了，但三元组之间的逻辑关系全是错的，等于建了一个“伪图谱”。

最后说一句：数据源分析不是一次性的活儿。随着业务变化，新数据源会不断冒出来，老数据源可能改结构或更换接口。保持你的数据源目录和配置文档持续更新，别等到管道挂了再去翻三个月的代码日志。