4. 数据采集与接入
在金融数据清洗流水线中，采集与接入是决定后续清洗质量与时效性的第一道关口。金融数据源天然具有异构、高频、强时序和口径不一的特点：行情接口按毫秒推送、财报文件按季度落地、关系库中的交易流水持续增长、实时流则要求低延迟消费。若采集层设计粗糙，后续无论怎样清洗都难以弥补数据缺失、重复或时间错位带来的问题。本节围绕四类主流数据源的读取方式、增量拉取策略与初始探查方法展开，帮助你建立可复用、可观测的接入层。
核心原理：四类数据源的接入特征
不同数据源的接入本质，是“拉”与“推”两种模式的组合。API 与数据库通常由客户端主动拉取，文件多为批量落地后读取，实时流则由服务端推送。理解其时间语义、幂等性和断点续传能力，是设计增量拉取的前提。
| 数据源 | 典型协议/工具 | 增量依据 | 主要风险 |
|---|---|---|---|
| REST API | HTTP/JSON、requests | 分页游标、时间戳参数 | 限流、字段漂移 |
| 数据库 | JDBC、SQLAlchemy | 自增 ID、更新时间戳 | 长事务、锁表 |
| 文件 | CSV、Parquet、SFTP | 文件修改时间、清单 | 重复投递、编码错误 |
| 实时流 | Kafka、WebSocket | Offset、水位线 | 乱序、重复消费 |
增量拉取的核心是“水位线”（watermark）管理：记录上一次成功拉取的最大时间戳或最大 ID，下次仅请求大于该值的数据。对于实时流，水位线体现为消费位点（offset），需配合幂等写入避免重复。
方法流程：从接入到初始探查
- 连接与认证：API 使用 或签名，数据库使用连接池，流使用消费者组。
- 增量拉取：按水位线构造请求，分页处理，失败重试并记录断点。
- 落地暂存：原始数据先写入 ODS 层，保留原始格式，便于回溯。
- 初始探查：统计行数、空值率、字段类型、时间范围与主键唯一性。
- 质量标记：对异常记录打标，进入后续清洗环节。
增量拉取的关键不是“拉多少”，而是“从哪里继续”。水位线必须持久化，且与业务时间解耦，避免因数据延迟导致漏拉。
案例/场景：行情 API 与交易库的增量接入
假设我们需要接入某行情 API 的日线数据，并同步交易库中的订单表。行情 API 支持按 start_date 与 end_date 查询，交易库订单表有自增主键 order_id 和更新时间 updated_at。
# 行情 API 增量拉取示例
import requests, json, datetime
watermark = load_watermark("market_daily") # 上次最大交易日
params = {
"symbol": "000001.SZ",
"start_date": watermark,
"end_date": datetime.date.today().isoformat(),
"": API_
}
resp = requests.get("https://api.example.com/daily", params=params, timeout=10)
data = resp.json()["data"]
save_raw(data, layer="ods_market_daily")
update_watermark("market_daily", max(d["trade_date"] for d in data))
-- 交易库增量拉取示例
SELECT order_id, user_id, amount, status, updated_at
FROM orders
WHERE updated_at > :last_watermark
ORDER BY updated_at ASC
LIMIT 5000;
实时流场景下，Kafka 消费者按 offset 拉取，需将 offset 提交与业务处理结果绑定，推荐使用“至少一次”加幂等去重。初始探查可先消费一小批样本，观察字段结构与时间分布。
误区与复盘
常见误区一：用业务时间做水位线。若数据存在延迟到达，按业务时间拉取会漏掉迟到记录。建议使用“入库时间”或“更新时间”作为水位线，并保留重叠窗口。
常见误区二：忽略分页与限流。API 通常有速率限制，未做退避重试会导致批量失败。应实现指数退避与最大重试次数。
复盘建议：每次接入新数据源后，先做一次全量探查，记录字段类型、空值率、主键唯一性和时间跨度，形成数据源档案。后续增量拉取时，定期比对水位线推进情况，发现停滞立即告警。
采集与接入不是一次性任务，而是持续运行的管道。把水位线、重试、幂等和探查固化到代码与配置中，才能让后续清洗环节稳定可靠。完成本节后，你应能为任意金融数据源设计出可断点续传、可观测、可回溯的接入方案。