3、数据采集与接入：数据源类型（行情、财务、另类）、实时采集（Kafka）与批量采集（Sqoop/DataX）、数据质量初检
数据采集，说白了就是「把数据从源头搬进仓库」的过程。这一步要是没做好，后面所有分析都是空中楼阁。我在因子工厂项目里踩过最大的坑，就是数据源没搞清楚就开始建表，结果返工了整整两周。
今天咱们就聊聊数据采集的三个核心问题：数据从哪来、怎么搬、搬完怎么验。
3.1 数据源类型：行情、财务、另类
因子工厂的数据源，我习惯分成三大类。你想想看，量化分析需要的信息，基本都跑不出这个圈。
| 数据源类型 | 典型数据 | 更新频率 | 数据量级 |
|---|---|---|---|
| 行情数据 | 股票Tick、K线、期货逐笔 | 毫秒~秒级 | TB级/天 |
| 财务数据 | 季报、年报、业绩预告 | 季度/月度 | GB级 |
| 另类数据 | 舆情、卫星图、电商销量 | 分钟~天级 | TB级 |
行情数据
这是因子工厂的「主食」。我记得刚接手项目时，行情数据一天就有200多GB。Tick级别的数据，每秒都在产生。这类数据的特点是：量大、实时性要求高、结构相对固定。比如上交所的Level-2行情，每3秒推送一次快照，包含买卖十档、逐笔成交等。
核心要点：行情数据一定要做「时间分区」+「标的物分区」。我见过有人把所有股票塞进一张表，查询时慢到怀疑人生。
财务数据
财务数据虽然量不大，但坑特别多。我曾经被「会计政策变更」坑过一次——某公司突然改了折旧方法，导致前后数据不可比。财务数据的特点：结构化强、更新频率低、但质量参差不齐。
嗯，这里要注意：财务数据往往有「修正值」。比如某公司发布一季报后，过两个月又发了个更正公告。你采集时如果只取第一次，后面分析就会出错。
另类数据
另类数据是因子工厂的「调味料」。比如爬取电商平台的销量数据、新闻舆情的情感分数、甚至卫星拍摄的停车场车辆数。这类数据的特点：格式多样、质量不可控、采集成本高。
我个人建议：另类数据先做小规模验证，确认有因子价值后再全量采集。否则很容易「采集一时爽，清洗火葬场」。
3.2 实时采集（Kafka）与批量采集（Sqoop/DataX）
数据源搞清楚了，接下来就是怎么搬。这里分两条路：实时流和批量批。
我的经验：不要为了「实时」而实时。行情数据必须实时，财务数据用批量就够了。我曾经见过有人把财务数据也用Kafka实时采集，结果每天就几条记录，纯属浪费资源。
实时采集：Kafka
Kafka在因子工厂里，主要扛行情数据。架构大概是这样的：
行情网关 → Kafka Producer → Kafka Cluster → Flink/Spark Streaming → 数据仓库为什么选Kafka？说白了，它就是个「超级消息队列」。行情数据每秒几万条，传统MQ根本扛不住。Kafka的吞吐量能到百万级/秒，而且数据持久化，不怕丢。
我记得有一次交易所行情延迟，Kafka里积压了2小时的数据。要是用别的中间件，早就OOM了。Kafka靠着磁盘顺序读写，硬是扛了下来。
配置上，我建议注意几个点：
- 分区数：按标的物哈希，保证同一只股票的数据有序
- 副本因子：至少2，防止节点挂了丢数据
- 消息大小：行情数据一般不超过1MB，默认配置够用
批量采集：Sqoop vs DataX
批量采集主要对付财务数据和历史行情。我常用的两个工具：Sqoop和DataX。
| 对比项 | Sqoop | DataX |
|---|---|---|
| 数据源 | 关系型数据库 → Hadoop | 多种数据源互导 |
| 性能 | MapReduce模式，适合大表 | 多线程模式，适合中规模 |
| 易用性 | 命令行，配置简单 | JSON配置，灵活度高 |
| 我的选择 | 财务数据（MySQL → Hive） | 历史行情（CSV → HDFS） |
举个例子，用Sqoop拉取财务数据：
sqoop import \
--connect jdbc:mysql://192.168.1.100:3306/finance \
--username root \
--password **** \
--table financial_report \
--where "report_date >= '2024-01-01'" \
--target-dir /user/hive/warehouse/finance \
--fields-terminated-by '\001' \
--num-mappers 4这里有个坑：--num-mappers不要设太大。我曾经设了16个并发，直接把源库的IO打满了，导致业务系统卡顿。后来学乖了，财务数据一般4~8个并发就够了。
DataX的配置更灵活，适合异构数据源。比如从CSV文件导入HDFS：
{
"job": {
"content": [{
"reader": {
"name": "txtfilereader",
"parameter": {
"path": ["/data/history/2024/*.csv"],
"encoding": "UTF-8",
"column": ["date", "code", "open", "high", "low", "close", "volume"]
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"path": "/user/hive/warehouse/history",
"fileName": "history_2024",
"writeMode": "append"
}
}
}]
}
}3.3 数据质量初检
数据搬进来了，不代表就能直接用。我吃过最大的亏就是：数据采集时没做质量检查，结果下游因子计算全用了脏数据。
数据质量初检，我总结为「三板斧」：
- 完整性检查：有没有缺失值？比如某只股票某天没有行情数据，是停牌了还是采集漏了？
- 一致性检查：字段格式对不对？比如日期字段，有的源是"2024-01-01"，有的是"20240101"，必须统一。
- 合理性检查：数值在不在合理范围？比如股票价格不可能为负数，成交量不可能为0（除非停牌）。
避坑指南：我曾经遇到过行情数据里出现「价格=999999」的异常值，原因是交易所的测试数据没过滤干净。如果没做合理性检查，这个异常值会直接污染所有因子。
我习惯在采集管道里加一个「质量检查层」。比如用Flink做实时检查：
// 伪代码示例
DataStream<Tick> tickStream = env.addSource(kafkaConsumer);
tickStream
.filter(tick -> tick.getPrice() > 0) // 价格必须大于0
.filter(tick -> tick.getVolume() >= 0) // 成交量不能为负
.filter(tick -> tick.getTimestamp() > 0) // 时间戳合法
.addSink(hbaseSink);对于批量数据，我建议在Sqoop/DataX导入后，跑一个简单的SQL做校验：
-- 检查财务数据完整性
SELECT
report_date,
COUNT(*) AS total_records,
SUM(CASE WHEN revenue IS NULL THEN 1 ELSE 0 END) AS null_revenue,
SUM(CASE WHEN profit IS NULL THEN 1 ELSE 0 END) AS null_profit
FROM financial_report
WHERE report_date = '2024-03-31'
GROUP BY report_date;如果发现异常数据，我一般会：
- 标记异常：在数据表里加一个
quality_flag字段，标记为'BAD' - 记录日志：把异常数据写入「数据质量日志表」，方便追溯
- 通知告警：如果异常比例超过阈值（比如1%），触发告警通知
我的习惯：数据质量初检不是一次性的。我会在每天凌晨跑一个「数据质量报告」，统计前一天所有数据源的质量指标。这样能尽早发现问题，而不是等到因子计算报错才去排查。
本章知识体系
下面这张图，概括了数据采集与接入的核心逻辑：
数据从源端出发，经过采集层（实时或批量），再到质量检查层，最终落入数据仓库。每一步都有坑，但每一步也都有成熟的解决方案。
好了，数据采集这块就聊到这儿。记住一句话：数据采集不是「搬砖」，而是「质检+搬运」的组合拳。搬得再快，质量不行也是白搭。