📚 课程目录第3章数据采集实战

3. 数据采集实战：日志采集方案设计、数据库增量采集（CDC）、API数据采集、采集任务监控与告警

数据采集是整个数据治理链条的起点。起点要是歪了，后面分析、建模、治理全白搭。我见过太多团队在采集阶段图省事，结果数据延迟、丢失、重复，最后花几倍精力去补坑。

这一章，咱们就聊聊四种最常见的采集场景：日志、数据库增量、API，以及怎么让采集任务稳定跑起来。

3.1 日志采集方案设计

日志采集，说白了就是把服务器、应用、中间件吐出来的日志，统一收拢到中央存储。最经典的组合就是 Filebeat + Kafka + Logstash。

3.1.1 架构选型：Filebeat vs Logstash vs Fluentd

组件	定位	资源消耗	适用场景
Filebeat	轻量采集器	极低（~10MB内存）	每台机器部署，只负责读文件、发数据
Logstash	数据处理管道	较高（~1GB内存）	集中节点做解析、过滤、转换
Fluentd	统一日志层	中等（~200MB内存）	需要插件生态、多输出目标

我个人习惯：采集端用 Filebeat，轻量无侵入；集中处理用 Logstash，灵活做 grok 解析；中间加一层 Kafka 做缓冲削峰。你想想看，如果 Logstash 挂了，数据还在 Kafka 里，不会丢。

3.1.2 关键配置：多行日志合并

Java 异常栈、SQL 慢查询这种多行日志，Filebeat 默认按行切分，会碎成好几条。我踩过这个坑——排查线上问题，异常栈被拆成 20 条记录，根本没法看。

# filebeat.yml
multiline.type: pattern
multiline.pattern: '^\d{4}-\d{2}-\d{2}'
multiline.negate: true
multiline.match: after

这段配置的意思是：如果当前行不是以日期开头，就合并到上一行末尾。嗯，这里要注意，pattern 要匹配你日志的实际时间格式。

小技巧： 先在测试环境用 filebeat -e -d "*" 跑一遍，看合并效果对不对。别直接上生产。

3.2 数据库增量采集（CDC）

CDC（Change Data Capture）就是捕获数据库的变更事件。传统做法是轮询时间戳或自增ID，但实时性差、对库有压力。现在主流是解析 binlog（MySQL）或 WAL（PostgreSQL）。

3.2.1 三种 CDC 方案对比

方案	原理	延迟	对源库影响	典型工具
基于查询	轮询时间戳/ID	秒~分钟级	有查询压力	自写脚本
基于日志解析	读取 binlog/WAL	毫秒级	几乎无影响	Canal、Debezium
基于触发器	数据库触发器写变更表	毫秒级	影响写性能	自建

我在项目中遇到过：用轮询方式采集订单表，结果业务高峰期数据库 CPU 飙到 90%。后来换成 Canal 解析 binlog，CPU 降到 5% 以下。说白了，日志解析才是 CDC 的正解。

3.2.2 Debezium + Kafka 实战配置

Debezium 是目前最成熟的 CDC 框架，基于 Kafka Connect 运行。下面是一个 MySQL 连接器的配置片段：

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium_pwd",
    "database.server.id": "223344",
    "database.server.name": "prod_orders",
    "database.include.list": "order_db",
    "table.include.list": "order_db.orders,order_db.order_items",
    "snapshot.mode": "initial",
    "offset.storage": "org.apache.kafka.connect.storage.FileOffsetBackingStore",
    "offset.storage.file.filename": "/data/offsets.dat",
    "offset.flush.interval.ms": "60000"
  }
}

注意： database.server.id 不能与 MySQL 集群中任何已有 server-id 重复。我曾经因为配了相同的 ID，导致主从同步异常，排查了一整天。

3.3 API 数据采集

外部系统、SaaS 平台的数据，通常只能通过 API 拉取。API 采集的核心挑战是：限流、分页、鉴权、重试。

3.3.1 通用采集框架设计

我建议把 API 采集抽象成三层：调度层、执行层、存储层。

调度层： 控制采集频率、并发数、重试策略。用 Quartz 或 XXL-Job 都行。
执行层： 处理鉴权、分页、数据转换。每个 API 写一个 Handler。
存储层： 原始数据落 Kafka 或对象存储，清洗后再入数仓。

3.3.2 分页与限流处理

大部分 API 用游标分页或偏移量分页。限流通常通过令牌桶实现。下面是一个 Python 示例：

import time
import requests
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=100, period=60)  # 每分钟最多100次
def fetch_page(url, params, headers):
    resp = requests.get(url, params=params, headers=headers, timeout=30)
    resp.raise_for_status()
    return resp.json()

def collect_all_pages(base_url, ):
    cursor = None
    all_data = []
    while True:
        params = {"cursor": cursor, "limit": 1000}
        headers = {"Authorization": f"Bearer {}"}
        data = fetch_page(base_url, params, headers)
        all_data.extend(data["items"])
        cursor = data.get("next_cursor")
        if not cursor:
            break
    return all_data

经验之谈： 限流参数别写死。我习惯把 calls 和 period 放到配置中心，方便根据 API 提供方的策略动态调整。

3.4 采集任务监控与告警

采集任务跑在凌晨，没人盯着。一旦挂了，早上才发现数据是空的。所以监控和告警必须前置。

3.4.1 核心监控指标

指标	含义	告警阈值
采集延迟	数据产生时间 vs 到达时间	> 5分钟
采集速率	每秒/每分钟采集条数	低于基线 50%
错误率	失败请求 / 总请求	> 1%
数据量波动	与历史同期对比	偏差 > 30%
Kafka Lag	消费者落后生产者的消息数	> 10000

3.4.2 告警分级与通知

别什么异常都发钉钉群。我见过一个团队，半夜因为网络抖动发了 200 条告警，第二天全员屏蔽了告警机器人。分级很重要：

P0（紧急）： 采集完全停止、数据丢失。电话 + 短信 + 群@所有人。
P1（严重）： 延迟超过阈值、错误率飙升。群消息 + 电话。
P2（警告）： 速率下降、数据量波动。群消息。
P3（通知）： 单次重试成功、临时限流。仅记录日志。

避坑指南： 我曾经把 Kafka Lag 告警阈值设成 1000，结果业务大促期间正常积压就超过 5000，告警刷屏。后来改成按基线动态阈值——取过去 7 天同一时刻的 P99 值，超过 1.5 倍才告警。

3.5 本章知识体系总览

下面这张图把四种采集方式、核心组件、监控链路串起来了。你可以把它当作本章的思维导图。

好了，这一章的内容就到这里。四种采集方式各有适用场景，但核心思路是一致的：采集要轻、缓冲要稳、监控要全。你可以在自己的项目中先挑一种场景落地，跑通之后再横向扩展。