3. 数据采集层：日志采集与消息队列选型

数据采集层，说白了就是风险监控系统的「耳朵」和「嘴巴」。耳朵负责听——从各个业务系统把日志捞上来；嘴巴负责说——把数据喂给下游处理引擎。这一层要是没搭好，后面再牛的算法也白搭。

我这些年做过的风险监控项目，踩坑最多的就是数据采集层。不是丢数据，就是数据格式乱七八糟，搞得下游解析逻辑改了一遍又一遍。今天咱们就把这块掰开揉碎了讲清楚。

3.1 日志采集工具：Filebeat vs Flume

选采集工具，我个人的习惯是先看场景。你想想看，如果你的业务系统全是容器化部署，日志都往标准输出打，那Filebeat就是天选之子。反过来，如果你们还在用老一套的Java应用，日志文件散落在各个服务器上，Flume可能更顺手。

核心区别一句话总结：

  • Filebeat：轻量、Go语言编写、资源占用极低，适合容器和云原生场景
  • Flume：Java生态、功能丰富、支持复杂的数据流拓扑，适合传统架构

我在项目中遇到过这么个事：有个金融客户，他们的交易日志每秒产生大概2万条，每条日志平均1.5KB。一开始他们用Flume，结果发现采集节点CPU飙到80%以上。后来换成Filebeat，CPU直接降到15%。为什么？Filebeat是Go写的，没有JVM那一套开销，而且它内部做了零拷贝优化。

Filebeat 配置示例

# filebeat.yml
filebeat.inputs:
- type: container
  paths:
    - /var/lib/docker/containers/*/*.log
  json.keys_under_root: true
  json.add_error_key: true

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "risk-log-topic"
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: gzip
  max_message_bytes: 10485760

嗯，这里要注意：required_acks 我建议设成1，别设成all。设成all虽然数据更安全，但吞吐量会掉一半。风险监控场景下，偶尔丢几条日志是可以接受的，但延迟高了会直接影响实时性。

Flume 配置示例

# flume.conf
agent.sources = tailSource
agent.channels = kafkaChannel
agent.sinks = kafkaSink

agent.sources.tailSource.type = spooldir
agent.sources.tailSource.spoolDir = /var/log/risk-app
agent.sources.tailSource.fileHeader = true
agent.sources.tailSource.batchSize = 1000

agent.channels.kafkaChannel.type = memory
agent.channels.kafkaChannel.capacity = 10000
agent.channels.kafkaChannel.transactionCapacity = 1000

agent.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkaSink.kafka.topic = risk-log-topic
agent.sinks.kafkaSink.kafka.bootstrap.servers = kafka1:9092,kafka2:9092
agent.sinks.kafkaSink.kafka.producer.acks = 1

我的经验之谈：Flume的channel选型很关键。memory channel快但会丢数据，file channel安全但慢。我一般建议用memory channel + 上游日志保留24小时，这样即使Flume挂了，还能从原始日志重新采集。

3.2 消息队列选型：为什么是Kafka？

你可能要问：消息队列那么多，RabbitMQ、RocketMQ、Pulsar，为什么风险监控系统几乎清一色选Kafka？

原因很简单：风险监控的核心诉求是「高吞吐」和「顺序性」。Kafka的日志结构存储，天生就是为这种场景设计的。我曾经做过压测，单机Kafka能扛住每秒50万条消息写入，延迟还在10ms以内。换成RabbitMQ，同样配置下5万条就开始丢消息了。

Kafka在风险监控中的关键配置：

  • 分区数：建议等于消费组内消费者数量的2-3倍，方便水平扩展
  • 副本因子：至少2，保证数据不丢
  • 日志保留时间：72小时，给数据回溯留足空间
  • 压缩方式：lz4或zstd，压缩比高且CPU开销小

Kafka 核心配置详解

# server.properties
# 分区数，根据业务量调整
num.partitions=12

# 日志保留72小时
log.retention.hours=72

# 每个分区日志最大1GB后滚动
log.segment.bytes=1073741824

# 副本因子
default.replication.factor=2

# 最小同步副本数
min.insync.replicas=1

# 消息最大大小，风险日志可能包含大payload
max.message.bytes=10485760

# 压缩类型
compression.type=lz4

我曾经踩过的坑：有次把 min.insync.replicas 设成了2，结果一台broker挂了之后，所有写入都报错。因为副本因子是2，最小同步副本也是2，那台broker一挂，就永远凑不齐2个同步副本了。正确的做法是：副本因子设3，最小同步副本设2，这样挂一台还能正常写。

3.3 数据格式标准化

这一节我觉得是最容易被忽视的。很多团队上来就搞采集、搞Kafka，结果数据格式五花八门——有的用JSON，有的用纯文本，有的甚至用自定义二进制。下游解析的时候，光适配格式就花了一半时间。

我建议统一用JSON格式，原因有三：

  1. 可读性强，调试方便
  2. 几乎所有语言都有原生支持
  3. schema可以灵活扩展，不影响老字段

标准日志格式定义

{
  "timestamp": "2024-01-15T10:30:00.123Z",
  "log_level": "ERROR",
  "source": {
    "service": "payment-service",
    "host": "10.0.1.23",
    "instance": "pod-payment-7f8b9c"
  },
  "event": {
    "type": "transaction_failed",
    "code": "ERR_TX_1001",
    "message": "支付超时，交易ID: TX202401151030001"
  },
  "context": {
    "user_id": "u_123456",
    "transaction_id": "TX202401151030001",
    "amount": 2999.00,
    "currency": "CNY",
    "risk_score": 0.85
  },
  "tags": ["high_risk", "payment", "timeout"],
  "extra": {}
}

标准化要点：

  • timestamp：必须用ISO 8601格式，带时区信息
  • source：记录来源服务、主机、实例，方便定位问题
  • event：事件类型和错误码，用于规则匹配
  • context：业务上下文，风险评分等关键信息
  • tags：标签系统，用于快速过滤和分类

你可能觉得这样定义太啰嗦。但我在项目中吃过亏——有一次线上出了个风险事件，日志里只有一条「交易失败」的消息，没有用户ID、没有交易金额，根本没法做根因分析。从那以后，我强制要求所有日志必须包含完整的context信息。

数据格式校验

光定义格式还不够，还得有校验机制。我一般会在采集层加一个轻量级的schema校验，用JSON Schema或者Avro都行。

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "required": ["timestamp", "log_level", "source", "event"],
  "properties": {
    "timestamp": {
      "type": "string",
      "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z$"
    },
    "log_level": {
      "type": "string",
      "enum": ["DEBUG", "INFO", "WARN", "ERROR", "FATAL"]
    },
    "source": {
      "type": "object",
      "required": ["service", "host"]
    },
    "event": {
      "type": "object",
      "required": ["type", "code"]
    }
  }
}

小技巧：校验不通过的日志不要直接丢弃，可以发到一个专门的「dead letter topic」里。我习惯保留7天，方便排查是哪个业务系统在乱打日志。等业务方修好了，再把这些数据重新处理一遍。

3.4 整体架构图

说了这么多，咱们用一张图把整个数据采集层的架构串起来。这张图我画了很多遍，每次给新团队培训都用它。

数据采集层架构图 数据源 应用服务器日志 容器标准输出 网络设备日志 采集层 Filebeat / Flume 格式校验 标准化处理 消息队列 Kafka 集群 分区: 12 副本: 2 消费层 实时计算引擎 规则引擎 告警服务 死信队列 (Dead Letter) 格式异常数据暂存 图例 数据源 采集层 消息队列 消费层 死信队列

这张图把整个数据流讲清楚了：从数据源出发，经过采集层做格式校验和标准化，然后进入Kafka消息队列，最后被消费层取走处理。如果格式校验不通过，数据会被打入死信队列，不会影响主流程。

总结一下本章核心要点：

  • 采集工具选型：容器场景用Filebeat，传统场景用Flume
  • 消息队列选型：高吞吐场景首选Kafka，注意分区和副本配置
  • 数据格式标准化：统一JSON格式，包含timestamp、source、event、context四个核心字段
  • 异常处理：格式校验不通过的数据打入死信队列，不要直接丢弃

数据采集层搭好了，后面的实时计算和规则引擎才能跑得顺畅。下一层咱们聊聊怎么用Flink做实时风险计算，那又是另一番天地了。

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

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