3. 数据采集层：日志采集、数据库采集与 API 采集设计

数据采集层是整个流水线的起点。说白了，源头要是脏了、慢了、丢了，后面做再多优化也是白搭。我见过太多团队在计算引擎上砸重金，结果采集端随便写个 shell 脚本去拉日志，最后数据对不上，排查了三天才发现是采集丢行。

这一章，我们聚焦三种最主流的采集场景：日志、数据库变更和API 拉取。每种场景我都会给出选型建议、配置要点，以及我踩过的坑。

3.1 日志采集：Filebeat vs Fluentd

日志采集是最常见的需求。应用打印的文本日志，需要实时或准实时地送入消息队列或存储系统。市面上两个主流工具：Filebeat 和 Fluentd。

我个人习惯这样选：如果只是把日志送到 Kafka 或 Elasticsearch，且对资源敏感，用 Filebeat。如果需要对日志做复杂的解析、过滤、路由，或者要同时输出到多个目标，用 Fluentd。

Filebeat 核心配置示例

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
    - /var/log/nginx/error.log
  multiline:
    pattern: '^\d{4}-\d{2}-\d{2}'
    negate: true
    match: after

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

嗯，这里要注意 required_acks 这个参数。设成 1 表示只要 leader 确认就返回成功，性能好但可能丢数据。设成 -1 表示所有副本都确认，可靠性最高但延迟大。我一般建议生产环境设成 1，配合 Kafka 的副本机制来保证不丢。

Fluentd 核心配置示例

<source>
  @type tail
  path /var/log/nginx/access.log
  pos_file /var/log/td-agent/nginx-access.pos
  tag nginx.access
  <parse>
    @type nginx
  </parse>
</source>

<match nginx.*>
  @type kafka2
  brokers kafka1:9092,kafka2:9092
  topic_key topic
  default_topic nginx-logs
  <buffer>
    @type file
    path /var/log/td-agent/buffer/nginx
    flush_interval 5s
    chunk_limit_size 8MB
  </buffer>
</match>

Fluentd 的 <buffer> 段是它的核心可靠性保障。数据先写入磁盘缓冲，再异步发送。如果 Kafka 挂了，数据不会丢，等恢复后自动重发。这个机制让我在几次 Kafka 集群故障中保住了数据。

3.2 数据库采集：CDC 与 Debezium

数据库采集，说白了就是捕获数据库的变更事件。传统做法是定时跑 SQL 查询增量数据，但这种方式延迟高、对数据库有压力。现在主流方案是 CDC（Change Data Capture），通过读取数据库的事务日志来捕获变更。

Debezium 是目前最流行的 CDC 框架。它基于 Kafka Connect，支持 MySQL、PostgreSQL、MongoDB、Oracle 等主流数据库。

Debezium MySQL 连接器配置示例

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-primary",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.include.list": "inventory",
    "table.include.list": "inventory.orders,inventory.customers",
    "database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "include.schema.changes": "true",
    "snapshot.mode": "initial"
  }
}

这里有个关键参数 snapshot.mode。设成 initial 表示启动时先做一次全量快照，然后切换到增量模式。如果表很大（比如上亿行），全量快照会持续很长时间，对数据库有压力。我建议在业务低峰期启动连接器，或者先用 when_needed 模式。

CDC 架构的可靠性设计

CDC 架构中，Kafka 是核心的缓冲层。Debezium 把变更事件写入 Kafka，下游的消费者从 Kafka 读取。这种解耦设计带来了几个好处：

• 异步解耦：数据库只负责写入 binlog，不关心下游消费速度
• 持久化缓冲：Kafka 可以保留大量消息，消费者可以回溯重放
• 多消费者：同一个变更事件可以被多个消费者消费，用于不同目的

但也要注意，CDC 不是银弹。它引入了额外的运维复杂度，Kafka 集群本身也需要高可用。如果团队没有 Kafka 运维经验，可以考虑使用 Debezium Server 直接输出到目标系统，跳过 Kafka。

3.3 API 采集设计

API 采集通常用于从外部系统拉取数据，比如第三方支付、CRM、广告平台等。这类采集的特点是：频率受限、数据量不确定、接口可能不稳定。

设计 API 采集时，我一般关注以下几个点：

1. 限流与重试：第三方 API 通常有 QPS 限制，需要做令牌桶或漏桶限流。重试策略要用指数退避，避免雪崩。
2. 增量与全量：优先使用增量接口（基于时间戳或游标），减少数据量。全量接口只用于初始化或数据修复。
3. 断点续传：记录每次拉取的位置（比如时间戳、页码），失败后从断点恢复，避免重复拉取大量数据。
4. 数据校验：拉取后做简单的校验，比如记录数、字段完整性、时间戳合理性。

API 采集框架示例（Python）

import time
import requests
from tenacity import retry, stop_after_attempt, wait_exponential

class APICollector:
    def __init__(self, base_url, , rate_limit=10):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {}"}
        self.rate_limit = rate_limit  # 每秒最大请求数
        self.last_request_time = 0

    def _rate_limit(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < 1.0 / self.rate_limit:
            time.sleep(1.0 / self.rate_limit - elapsed)
        self.last_request_time = time.time()

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    def _request(self, endpoint, params=None):
        self._rate_limit()
        response = requests.get(
            f"{self.base_url}/{endpoint}",
            headers=self.headers,
            params=params,
            timeout=30
        )
        response.raise_for_status()
        return response.json()

    def collect_incremental(self, endpoint, since_timestamp, page_size=100):
        page = 1
        all_records = []
        while True:
            params = {
                "since": since_timestamp,
                "page": page,
                "per_page": page_size
            }
            data = self._request(endpoint, params)
            records = data.get("data", [])
            if not records:
                break
            all_records.extend(records)
            page += 1
        return all_records

这个示例里用了 tenacity 库做重试，指数退避策略。第一次重试等 2 秒，第二次等 4 秒，第三次等 8 秒。如果三次都失败，就抛出异常，由上层决定是告警还是跳过。

3.4 三种采集方式的对比与选型

你想想看，如果业务要求秒级延迟，API 采集肯定不行。如果数据一致性要求极高（比如金融交易），日志采集可能丢行，也不合适。CDC 是最佳选择。但如果源系统是第三方，你拿不到数据库权限，那就只能用 API 采集。

3.5 数据采集层架构总览

下面这张图展示了三种采集方式在整体架构中的位置和关系。

这张图展示了数据从源系统到采集组件，再到缓冲层的完整路径。三种采集方式各有侧重，但最终都汇聚到 Kafka 这个统一的缓冲层。这样做的好处是：下游的流处理、数据湖、实时分析等组件，只需要对接 Kafka 一个入口，不需要关心上游是哪种采集方式。