数据采集与存储：常见数据源、采集工具与存储格式

大家好，我是你们的老朋友。今天我们来聊聊数据回放里最基础、也最容易被忽视的一环——数据采集与存储。

说实话，我见过太多团队在回放时翻车，最后发现根因不是算法不对，而是数据源没接对，或者存储格式选错了。嗯，这期内容就是帮你把这些坑提前填上。

常见数据源：日志、数据库、消息队列

数据回放的第一步，是搞清楚数据从哪来。我个人习惯把数据源分成三类：日志、数据库、消息队列。这三类几乎覆盖了90%以上的生产场景。

1. 日志文件

日志是最原始、最通用的数据源。你想想看，不管是Nginx的访问日志，还是Java应用的log4j输出，本质上都是文本流。

我在项目中遇到过一个问题：某次回放时发现数据量对不上，排查了半天，原来是日志文件被logrotate切走了，采集工具读了个空文件。所以，处理日志源时，一定要关注文件滚动策略。

  核心要点：日志采集需要支持断点续传、文件监听、滚动处理。

2. 数据库

数据库作为数据源，常见于业务系统的变更捕获。比如MySQL的binlog、PostgreSQL的WAL日志。

为什么用数据库？因为日志里可能没有完整的业务上下文。举个例子，用户下单的完整链路，日志可能只记录了「下单成功」，但数据库里有订单金额、商品ID、用户地址等完整信息。

注意：直接从数据库拉取全量数据做回放，容易对线上库造成压力。我曾经见过一个团队，回放时直接select * from orders，结果把生产库的IO打满了。建议使用CDC（Change Data Capture）方式，只拉取增量变更。

3. 消息队列

消息队列是现在微服务架构下的标配。Kafka、RocketMQ、RabbitMQ，这些中间件里流转的数据，天然就是流式数据，非常适合做回放。

我个人最喜欢用Kafka做数据源，因为它有offset机制，可以精确控制回放的位置。你想回放昨天的数据？直接把consumer的offset重置到昨天就行。

数据采集工具：Flume vs Logstash

数据源搞定了，接下来就是怎么把数据「搬」到存储系统里。这里我重点讲两个工具：Flume和Logstash。

特性	Flume	Logstash
开发语言	Java	Ruby（JRuby）
核心模型	Source-Channel-Sink	Input-Filter-Output
数据可靠性	支持事务，可靠性高	默认至少一次，可配置
插件生态	中等，Hadoop生态友好	丰富，ELK全家桶
适用场景	大数据离线采集	实时日志处理

怎么选？我个人的经验是：

如果你的下游是HDFS、Hive这类大数据组件，用Flume。它和Hadoop生态的集成做得很好。
如果你的下游是Elasticsearch，或者你需要对数据做复杂的过滤、转换，用Logstash。它的filter插件非常强大。

我曾经在项目中同时用过这两个工具。当时的情况是：日志先通过Flume采集到Kafka，然后Logstash从Kafka消费，做字段解析和清洗，最后写入ES。说白了，各司其职。

Flume配置示例

# 一个简单的Flume配置：监听日志目录，写入HDFS
agent.sources = tailSource
agent.channels = fileChannel
agent.sinks = hdfsSink

agent.sources.tailSource.type = spooldir
agent.sources.tailSource.spoolDir = /var/log/app

agent.channels.fileChannel.type = file
agent.channels.fileChannel.checkpointDir = /tmp/flume/checkpoint

agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsSink.hdfs.path = /data/logs/%Y%m%d

Logstash配置示例

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["app-logs"]
    codec => "json"
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:msg}" }
  }
  date {
    match => ["timestamp", "ISO8601"]
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
  }
}

避坑指南：我曾经在Logstash里用了一个性能很差的grok正则，结果CPU直接飙到90%。建议能用json解析就别用grok，实在要用，先测试正则效率。

存储格式：CSV、Parquet、Avro

数据采过来了，存成什么格式？这直接决定了回放的效率和存储成本。

1. CSV

CSV是最简单的格式，人类可读，Excel能直接打开。但它的缺点也很明显：没有schema信息、不支持嵌套结构、压缩率低。

我一般只在数据量小（几百MB以内）或者需要人工审查的场景下用CSV。生产环境？算了吧。

2. Parquet

Parquet是列式存储格式，专为分析型查询优化。它的特点是：

按列存储，查询时只读取需要的列
支持复杂的嵌套数据结构
压缩率高，通常能比CSV节省70%以上的空间
自带schema信息

我在做数据回放时，如果数据量超过10GB，首选Parquet。尤其是回放场景下，我们经常只需要某几列数据，列式存储的优势就体现出来了。

3. Avro

Avro是行式存储格式，但它的亮点在于：

数据文件自带schema，读写双方不需要额外约定格式
支持schema演化，比如新增字段不会破坏旧数据
序列化后的数据体积小，适合网络传输

Avro特别适合Kafka这类消息队列的场景。你想想看，生产者写Avro数据，消费者读Avro数据，只要schema兼容，中间随便改字段都没问题。

格式	存储方式	压缩率	Schema支持	适用场景
CSV	行式	低	无	小数据量、人工查看
Parquet	列式	高	有	分析查询、大数据回放
Avro	行式	中	有（支持演化）	消息队列、数据交换

  我的建议：如果数据需要频繁写入和读取（比如实时流），用Avro。如果数据主要是批量分析（比如离线回放），用Parquet。CSV？嗯，尽量别用。

知识体系总览

下面这张图，是我自己梳理的数据采集与存储的核心逻辑。你可以把它当作一张「地图」，遇到问题时回来看看。

好了，这一章的内容就到这里。数据采集和存储是数据回放的「地基」，地基不稳，后面再好的算法也白搭。希望今天的内容能帮你少走一些弯路。

最后提醒一句：不管选什么工具、什么格式，一定要在生产环境做压力测试。我曾经在Flume的channel上吃过亏，默认配置扛不住突发流量，数据丢了一堆。嗯，从那以后，我上线前必做压测。

可可精品资料库

📚 课程目录

数据采集与存储：常见数据源、采集工具与存储格式

常见数据源：日志、数据库、消息队列

1. 日志文件

2. 数据库

3. 消息队列

数据采集工具：Flume vs Logstash

Flume配置示例

Logstash配置示例

存储格式：CSV、Parquet、Avro

1. CSV

2. Parquet

3. Avro

知识体系总览