数据源接入：结构化、半结构化与非结构化

数据源接入，说白了就是搞清楚你的数据从哪来、长什么样、怎么把它拽进 ELT 管道。我见过不少团队，一上来就急着写代码，结果被数据源的「脾气」折腾得够呛。这一章咱们就把三类数据源——结构化、半结构化、非结构化——挨个捋一遍。

结构化数据源：MySQL 与 PostgreSQL

结构化数据源，就是那种规规矩矩的表。行是行，列是列，主键、外键、索引一应俱全。MySQL 和 PostgreSQL 是咱们最常打交道的两个。

MySQL 接入要点

我个人习惯用 批量读取 + 增量同步 的方式接 MySQL。全量同步？那是小表才敢干的事。一张几千万行的订单表，你每周全量拉一次，管道迟早会崩。

举个例子，假设订单表有个 update_time 字段：

-- 增量抽取 SQL
SELECT *
FROM orders
WHERE update_time > '2024-01-01 00:00:00'
  AND update_time <= '2024-01-02 00:00:00'
ORDER BY id;

嗯，这里要注意：时间字段一定要有索引。我在项目中遇到过，某次没加索引，全表扫描直接把源库 CPU 打满，业务方差点打电话过来骂人。

PostgreSQL 接入要点

PostgreSQL 比 MySQL 多了一些「高级玩具」，比如 逻辑复制 和 WAL 日志。如果你需要近乎实时的数据同步，逻辑复制是首选。

特性	MySQL	PostgreSQL
增量方式	binlog / 自增 ID / 时间戳	逻辑复制 / WAL / 时间戳
批量读取	支持，但大表需分页	支持，游标更稳定
数据类型	基础类型为主	支持 JSON、数组、自定义类型
连接池	推荐 HikariCP	推荐 PgBouncer

你想想看，如果源表里存了 JSON 字段，PostgreSQL 可以直接在 SQL 里解析，省掉一层后处理。MySQL 8.0 虽然也支持 JSON，但性能上还是差一截。

半结构化数据源：JSON 与 XML

半结构化数据，说白了就是「有结构但不那么死板」。JSON 和 XML 是最常见的两种。它们不像数据库表那样有固定的 schema，但也不是纯文本。

JSON 接入

JSON 现在几乎是 API 接口的标配。我建议在 ELT 的 Extract 阶段就做一次 扁平化，把嵌套的 JSON 拍平成宽表。

举个例子，一个典型的用户行为日志：

{
  "user_id": 12345,
  "event": "click",
  "timestamp": "2024-01-15T10:30:00Z",
  "properties": {
    "page": "home",
    "button": "signup",
    "duration_ms": 2500
  }
}

扁平化之后，变成这样：

| user_id | event | timestamp           | page | button | duration_ms |
|---------|-------|---------------------|------|--------|-------------|
| 12345   | click | 2024-01-15 10:30:00 | home | signup | 2500        |

为什么要扁平化？因为大多数分析引擎（比如 ClickHouse、BigQuery）对宽表的查询效率远高于嵌套结构。我曾经接手过一个项目，原始 JSON 嵌套了 5 层，ETL 跑一次要 40 分钟。扁平化之后，15 分钟搞定。

XML 接入

XML 现在用得少了，但在金融、医疗、政府系统里还是「老大哥」。XML 的解析比 JSON 重得多，因为它有命名空间、属性、CDATA 这些概念。

我个人习惯用 SAX 解析 而不是 DOM 解析。DOM 会把整个 XML 文档加载到内存，一个 500MB 的 XML 文件就能把 JVM 撑爆。SAX 是流式解析，边读边处理，内存占用小得多。

# Python 示例：使用 SAX 解析 XML
import xml.sax

class OrderHandler(xml.sax.ContentHandler):
    def startElement(self, name, attrs):
        if name == "order":
            self.current_order = {}
            self.current_order["id"] = attrs.get("id")

    def characters(self, content):
        if self.current_tag:
            self.current_order[self.current_tag] = content.strip()

    def endElement(self, name):
        if name == "order":
            # 写入目标表
            pass

你想想看，XML 的 schema 定义（XSD）其实是个好东西。它告诉你哪些字段是必填的、什么类型、取值范围。我建议在接入 XML 时，先拿 XSD 校验一遍，避免脏数据进管道。

非结构化数据源：日志文件与图片

非结构化数据，说白了就是「没固定格式」的数据。日志文件、图片、视频、PDF……这类数据没法直接塞进关系型数据库，得先做一层处理。

日志文件接入

日志文件是 ELT 里最常见的非结构化数据。Nginx 日志、应用日志、系统日志……每秒钟可能产生几百 MB。

我建议用 Filebeat + Kafka 的组合来收日志。Filebeat 负责采集和轻量处理，Kafka 负责缓冲和削峰。ELT 管道从 Kafka 消费数据，再做解析和结构化。

举个例子，一条 Nginx 日志：

192.168.1.1 - - [15/Jan/2024:10:30:00 +0800] "GET /api/users HTTP/1.1" 200 1234

解析后：

| ip            | timestamp           | method | path       | status | size |
|---------------|---------------------|--------|------------|--------|------|
| 192.168.1.1   | 2024-01-15 10:30:00 | GET    | /api/users | 200    | 1234 |

嗯，这里要注意：日志格式可能随时变。我曾经遇到过，开发团队在日志里加了一个字段，没通知我们，结果解析器直接报错，管道停了两个小时。从那以后，我每次上线前都会加一个 schema 校验，格式变了就报警。

图片接入

图片接入 ELT 管道，通常不是为了分析图片本身，而是为了提取元数据或做后续的 AI 处理。

我建议的做法是：

1. 存储路径：图片本身存在对象存储（S3、OSS）里，ELT 管道只存路径。
2. 元数据提取：用 EXIF 工具提取拍摄时间、GPS 坐标、设备型号等信息。
3. 特征向量：如果需要做图像识别，可以在 Extract 阶段调用预训练模型，生成特征向量，存入向量数据库。

三类数据源接入对比

维度	结构化	半结构化	非结构化
典型代表	MySQL, PostgreSQL	JSON, XML	日志文件, 图片
接入方式	JDBC/ODBC 批量读取	解析器 + 扁平化	采集代理 + 对象存储
增量策略	binlog / 时间戳	全量 + 增量标记	文件监听 + 偏移量
Schema 管理	固定，由 DDL 定义	灵活，需运行时推断	无，需提取阶段定义
常见坑点	大表全量扫描	嵌套层级过深	格式变化无通知

数据源接入的整体流程

下面这张图，是我自己画的数据源接入流程。你可以看到，不管数据源长什么样，最终都要经过 采集 → 解析 → 校验 → 暂存 这四个步骤，才能进入 ELT 的核心管道。

你看，不管数据源是 MySQL 还是日志文件，流程骨架是一样的。区别只在于采集和解析的具体实现。把这个流程想清楚，后面写代码的时候就不会跑偏。

好了，数据源接入这块就聊到这儿。记住一句话：接入阶段多花点心思，后面清洗和转换阶段能省一半的力气。