日志采集与传输：从源头到管道，一个都不能少

大家好，我是你们的老朋友。今天咱们聊聊日志采集与传输这块。

说实话，很多风控系统最后出问题，不是规则写得不好，也不是模型不够准，而是——日志丢了。嗯，你没听错。日志丢了，数据不全，风控再强也是白搭。所以这一章，我重点讲三个核心组件：Filebeat、Logstash、Kafka。它们怎么配合，怎么保证数据不丢，我踩过的坑也会一并分享。

1. Filebeat：轻量级采集器，我为什么偏爱它

先说说Filebeat。你可能听过Logstash也能采集日志，但为什么我还要用Filebeat？

原因很简单：轻。 Filebeat是用Go写的，资源占用极低。我曾在生产环境做过对比——同样采集1000个日志文件，Filebeat只占不到50MB内存，而Logstash直接飙到500MB+。你想想看，如果每台机器都跑Logstash，那成本得多高？

Filebeat的核心原理其实不复杂：

  • 监听器（Harvester）：每个日志文件对应一个harvester，逐行读取内容
  • 输入（Input）：支持多种来源，比如文件、stdin、TCP/UDP
  • 输出（Output）：可以发给Elasticsearch、Logstash、Kafka等
  • 内部队列（Spool）：暂存数据，防止突发流量打崩下游

我个人习惯用Filebeat + Kafka的组合。为什么？因为Filebeat直接写Kafka，少了一层Logstash转发，延迟更低。但如果你需要复杂的数据清洗，那还是得走Logstash。

避坑指南： 我曾经遇到过Filebeat采集日志时，因为日志文件轮转（rotation）太快，导致harvester还没读完就被删了，数据直接丢失。后来我加了 close_inactive: 5m 和 clean_removed: false 才解决。嗯，这里要注意，日志轮转策略要和Filebeat配置对齐。

下面是一个典型的Filebeat配置示例：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
  fields:
    app: nginx
    env: prod

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "nginx-logs"
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

看到 required_acks: 1 了吗？这表示只要Kafka leader确认收到就算成功。如果追求更高可靠性，可以改成 -1（所有副本都确认），但性能会下降。我一般用1，够用了。

2. Logstash：管道工，也是数据清洗工

Logstash，说白了就是一条数据管道。它从各种源头拉数据，做过滤、转换、格式化，再吐出去。我见过很多团队只用Logstash做采集，结果内存爆了——其实它更适合做中间处理，而不是边缘采集。

Logstash的三大核心组件：

  • Input：接收数据，支持Filebeat、Kafka、TCP等
  • Filter：数据清洗，比如解析JSON、正则提取、GeoIP定位
  • Output：输出到Elasticsearch、Kafka、S3等

举个例子，我们风控系统经常要解析Nginx日志，提取IP、时间、URL、状态码。用Logstash的grok插件，一行正则搞定：

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
  geoip {
    source => "clientip"
  }
}

你看，几行配置就把原始日志变成了结构化数据。但要注意——grok性能开销不小。我曾在生产环境遇到Logstash处理速度跟不上，导致Kafka积压。后来优化了正则表达式，把不必要字段去掉，才缓过来。

我的经验： Logstash的filter尽量精简。如果数据量超过每秒10万条，建议把部分清洗逻辑放到消费端（比如Spark Streaming）去做，别全压在Logstash上。

3. Kafka：消息队列，风控数据的生命线

Kafka在风控系统里扮演什么角色？说白了，就是缓冲+解耦。没有Kafka，采集端和处理端直接耦合，一旦处理端挂了，数据就丢了。有了Kafka，数据先存起来，消费端慢慢处理。

Kafka的几个关键概念，我简单捋一下：

概念 说明 我的建议
Topic 消息分类，比如"nginx-logs"、"order-events" 按业务类型分topic，别混在一起
Partition 分区，实现并行消费 分区数=消费线程数，别设太多
Offset 消息偏移量，记录消费位置 自动提交有风险，建议手动提交
Replication 副本数，保证数据不丢 至少2个副本，生产环境建议3个

我记得有一次线上事故，就是因为Kafka分区数设置不合理。当时我们一个topic有50个分区，但消费端只有5个线程，结果大部分分区都没人消费，数据积压了整整2小时。后来我把分区数调成10，消费线程也调到10，问题才解决。

警告： Kafka的 acks 参数一定要根据业务场景设置。风控数据对可靠性要求高，我建议用 acks=all（即-1），配合 min.insync.replicas=2，确保数据不丢。但如果你追求吞吐量，可以适当降低要求。

4. 日志传输可靠性保障：我踩过的坑，你别再踩

日志传输的可靠性，说白了就是三个问题：不丢、不重、不乱序。我一个个说。

4.1 不丢数据

这是最基本的要求。Filebeat有内部队列，默认是 queue.mem.events: 4096，如果下游Kafka挂了，数据会暂存在内存里。但内存有限，满了怎么办？Filebeat会阻塞采集，直到队列有空位。这其实是一种背压机制——宁可慢，不能丢。

我曾经犯过一个错误：把 queue.mem.events 设得太大（比如100万），结果内存爆了，Filebeat直接OOM。后来我改成 queue.mem.events: 2048，配合 queue.mem.flush.min_events: 512，既保证了吞吐，又控制了内存。

4.2 不重复数据

重复数据比丢数据更恶心。风控系统里，重复的订单事件可能导致重复扣款，那可就麻烦了。

Kafka的 enable.idempotence=true 可以保证生产者发送的消息不重复。但注意，这需要配合 acks=all 使用。我建议风控场景下，消费端也要做幂等处理——比如用唯一ID去重。

4.3 不乱序

乱序问题在风控里很常见。比如用户先下单后支付，但支付日志先到，下单日志后到，风控规则就可能误判。

解决方案：Kafka同一个分区内是有序的，所以把同一用户的事件发到同一个分区就行。用 keyed message，key设为用户ID：

producer.send(new ProducerRecord<>("order-events", userId, event));

这样，同一个用户的事件永远在一个分区，顺序就保证了。

总结一下我的经验： 日志传输的可靠性，不是靠某一个组件，而是靠整个链路的设计。Filebeat负责采集不丢，Kafka负责存储不丢，消费端负责处理不重。每一层都要有兜底机制。

5. 整体架构图：一张图看懂采集与传输

下面我用一张SVG图，把Filebeat、Logstash、Kafka的协作关系画出来。你看完应该就明白了。

日志采集与传输架构图 应用服务器 Nginx/Java/Node Filebeat 轻量采集器 Kafka 集群 Topic: nginx-logs Partition 0/1/2 Logstash 过滤/解析/转换 Elasticsearch 存储与检索 风控消费端 Spark/Flink/Java 图例： 应用 采集 消息队列 处理 存储 消费 说明：Filebeat采集日志 → Kafka缓冲 → Logstash清洗 → ES存储 / 风控消费

你看，整个链路很清晰：应用服务器产生日志，Filebeat采集后发到Kafka，Kafka作为缓冲层，下游可以接Logstash做清洗，也可以直接给风控消费端。这样即使消费端挂了，数据还在Kafka里，不会丢。

一个小技巧： 如果你们公司日志量特别大（比如每天几十TB），可以考虑用Filebeat直接写S3或HDFS归档，Kafka只保留最近7天的数据。这样既省钱，又保证实时性。

好了，这一章就到这里。日志采集与传输是风控系统的地基，地基不稳，上层再漂亮也没用。希望我踩过的坑，能帮你少走弯路。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →