数据源统一接入：Kafka 作为统一数据管道

好，咱们进入第二章。上一章聊了流批一体的整体架构思路，这一章我打算把最底层、也是最关键的一环讲透——数据源怎么统一接进来。

你想想看，一个典型的数据平台，数据源可能有几十种：MySQL 的 binlog、应用程序的日志、前端埋点、IoT 设备上报、第三方 API 回调……如果每种数据源都单独写一套接入逻辑，那维护成本简直不敢想。我早期带团队做实时数仓时，就吃过这个亏——接了 6 个数据源，用了 4 种不同的采集方式，最后光排查数据丢失问题就花了两周。

后来我们统一用 Kafka 做数据管道，才算是把这个问题彻底解决了。

为什么是 Kafka？

说白了，Kafka 在流批一体架构里扮演的角色，就是数据总线。所有数据先进 Kafka，下游的流处理、批处理、OLAP 引擎都从 Kafka 消费。这样做有几个明显的好处：

  • 解耦：数据生产者和消费者互不感知，各自独立演进。
  • 缓冲：生产端突发流量不会直接打垮下游，Kafka 的持久化能力扛得住。
  • 回溯：数据在 Kafka 里保留一段时间，批处理可以重读历史数据，流处理也能回溯重算。
  • 多订阅：一份数据可以被多个消费者以不同方式消费——流作业做实时聚合，批作业做 T+1 全量分析，互不干扰。

核心原则：在流批一体架构中，Kafka 是唯一的数据入口。所有数据源必须经过 Kafka 才能进入计算层。这条规则一旦打破，数据一致性就很难保证了。

Kafka Topic 设计原则

Topic 是 Kafka 里最基础的概念，但设计得好不好，直接影响整个管线的性能和可维护性。我见过不少团队，Topic 命名随意、分区数拍脑袋定、数据混在一个 Topic 里……结果后面运维起来非常痛苦。

我个人习惯遵循以下几个原则：

1. 命名规范：领域 + 数据源 + 事件类型

一个好的 Topic 命名，应该让人一眼就能看出它承载的是什么数据。我推荐这种格式：

<domain>.<source>.<event_type>.<version>

举个例子：

  • order.mysql.binlog.v1 —— 订单库的 binlog
  • user.app.click.v2 —— 用户点击事件
  • payment.thirdparty.callback.v1 —— 支付回调

这样命名，你在 Kafka Manager 或者命令行里一看就明白。我曾经接手过一个项目，Topic 名叫 data_topic_001、test_2……嗯，那感觉就像在垃圾堆里找东西。

2. 分区数：不是越多越好

分区数决定了 Kafka 的并行度。但很多人有个误区——分区越多吞吐越高。其实不然。

分区数过多会带来几个问题：

  • 文件句柄数飙升，操作系统扛不住
  • Leader 选举时间变长
  • 客户端内存开销增大

我一般这样估算：

因素 建议
目标吞吐量 分区数 ≥ 目标吞吐量 / 单个分区最大吞吐
消费者并行度 分区数 = 消费者线程数（或整数倍）
数据有序性要求 需要严格有序的场景，分区数不宜过多
实践经验 单个 Topic 分区数建议不超过 100

我的经验：如果拿不准，先设 12~24 个分区。后续可以通过 Kafka 的 kafka-reassign-partitions.sh 工具动态调整，但这个过程会触发数据重分布，生产环境要谨慎操作。

3. 数据隔离：不同数据类型分不同 Topic

千万别把订单数据和日志数据混在一个 Topic 里。原因很简单：

  • 它们的生命周期不同——日志可能只需要保留 3 天，订单数据要保留 30 天
  • 它们的消费方式不同——订单数据需要精确一次语义，日志可以容忍丢几条
  • 它们的 Schema 不同——混在一起会导致 Schema 管理复杂化

我见过一个案例，有人把所有数据都塞进一个 Topic，然后用消息里的 type 字段区分。结果下游消费者要过滤掉 90% 的无关数据，白白浪费了磁盘和网络带宽。

Schema Registry 与 Avro 序列化

数据进了 Kafka，怎么保证生产者和消费者对数据格式的理解一致？

你可能会说，用 JSON 不就行了？嗯，JSON 确实方便，但有几个硬伤：

  • 体积大：同样的数据，JSON 比 Avro 大 3~5 倍
  • 无 Schema 约束：生产端改了字段名，消费端可能直接挂掉
  • 解析慢：JSON 解析需要遍历整个字符串

所以，在流批一体架构里，我强烈推荐使用 Avro + Schema Registry 的组合。

Avro 是什么？

Avro 是一种二进制序列化格式，由 Hadoop 生态提出。它的核心特点是：

  • Schema 与数据分离：数据本身不携带 Schema，而是通过 Schema ID 关联
  • 支持 Schema 演化：可以添加字段、删除字段、修改类型（兼容模式下）
  • 压缩率高：二进制格式，体积小，序列化/反序列化速度快

Schema Registry 的作用

Schema Registry 是一个独立的服务，负责存储和管理 Avro Schema。生产者在发送数据前，先向 Registry 注册 Schema，拿到一个 Schema ID；然后把 Schema ID 塞进消息头，数据体用 Avro 二进制编码。消费者收到消息后，从消息头取出 Schema ID，去 Registry 拉取对应的 Schema，再用它来反序列化。

这样做的好处是：

  • 数据体积最小化：消息里只带一个 int 类型的 Schema ID，而不是完整的 Schema 字符串
  • 兼容性检查：Registry 可以配置兼容性策略（BACKWARD、FORWARD、FULL 等），防止不兼容的 Schema 变更上线
  • 集中管理：所有 Schema 都在一个地方，方便审计和版本回溯

注意：Schema Registry 是流批一体架构中的关键依赖，必须高可用部署。我曾经遇到过一次 Registry 宕机，导致所有生产者和消费者都无法注册或拉取 Schema，整个数据管道瘫痪了 40 分钟。所以，至少部署 3 个节点，并且做好监控告警。

Avro Schema 示例

下面是一个订单事件的 Avro Schema 定义：

{
  "namespace": "com.example.order",
  "type": "record",
  "name": "OrderEvent",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "user_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": {"type": "enum", "name": "OrderStatus", "symbols": ["CREATED", "PAID", "SHIPPED", "COMPLETED", "CANCELLED"]}},
    {"name": "created_at", "type": "long", "logicalType": "timestamp-millis"},
    {"name": "items", "type": {"type": "array", "items": {
      "type": "record",
      "name": "OrderItem",
      "fields": [
        {"name": "product_id", "type": "string"},
        {"name": "quantity", "type": "int"},
        {"name": "price", "type": "double"}
      ]
    }}}
  ]
}

这个 Schema 定义了一个订单事件，包含订单 ID、用户 ID、金额、状态、创建时间和商品列表。注意 logicalType 字段，它告诉 Avro 这个 long 类型实际上是一个毫秒级时间戳，序列化/反序列化时会自动转换。

多源数据接入实战

理论讲完了，咱们看看实际怎么接入。我以三种常见的数据源为例：

1. MySQL binlog 接入

MySQL 的 binlog 记录了所有数据变更，是实时数仓最常用的数据源之一。我们一般用 Canal 或 Debezium 来捕获 binlog 并写入 Kafka。

以 Debezium 为例，配置大概是这样：

# connector 配置
name=order-connector
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=184054
database.server.name=order_server
database.include.list=order_db
table.include.list=order_db.orders,order_db.order_items
topic.prefix=order
snapshot.mode=initial
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState

这里有几个关键点：

  • database.server.name 和 topic.prefix 共同决定了 Topic 名称，比如 order.order_db.orders
  • snapshot.mode=initial 表示先做一次全量快照，再增量同步
  • transforms.unwrap 会把 Debezium 的复杂消息结构简化，只保留变更后的数据

避坑指南：我曾经遇到过 binlog 接入导致 MySQL 主库压力过大。原因是 Debezium 在 snapshot 阶段会扫描全表，如果表很大（比如上亿行），可能会把主库的 IO 打满。解决办法是：从库读取 binlog，或者用 snapshot.mode=when_needed 配合手动控制快照时机。

2. 应用日志接入

应用日志通常通过 Filebeat 或 Fluentd 采集，然后写入 Kafka。以 Filebeat 为例：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  fields:
    app_name: user-service
    env: production
  fields_under_root: true

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
  topic: 'app.log.user-service'
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

注意这里我用了 fields 来标记应用名称和环境，这样在 Kafka 消费端可以方便地过滤和路由。另外，compression: gzip 可以显著减少网络带宽消耗——日志数据通常有很高的重复性，压缩比能达到 5:1 甚至更高。

3. 前端埋点接入

前端埋点数据的特点是：量大、字段多、允许少量丢失。我们一般用 Nginx + Lua 或者 API Gateway 接收埋点请求，然后批量写入 Kafka。

一个简单的 Nginx 配置示例：

location /track {
    content_by_lua_block {
        local cjson = require "cjson"
        local producer = require "resty.kafka.producer"
        
        local broker_list = {
            { host = "kafka1", port = 9092 },
            { host = "kafka2", port = 9092 }
        }
        
        local p = producer:new(broker_list, { producer_type = "async" })
        
        ngx.req.read_body()
        local data = ngx.req.get_body_data()
        
        local ok, err = p:send("track.click", nil, data)
        if not ok then
            ngx.log(ngx.ERR, "kafka send failed: ", err)
        end
        
        ngx.say("ok")
    }
}

这里用了异步生产者（producer_type = "async"），前端请求不会阻塞等待 Kafka 确认，适合高并发场景。但要注意，异步模式下如果 Kafka 宕机，数据可能会丢失——嗯，这就是前面说的「允许少量丢失」的权衡。

整体架构图

下面这张图展示了多源数据统一接入 Kafka 的整体流程：

多源数据统一接入 Kafka 架构图 MySQL binlog Debezium / Canal 应用日志 Filebeat / Fluentd 前端埋点 Nginx + Lua / API Gateway Kafka 集群（统一数据管道） Topic: order.mysql.binlog.v1 Topic: app.log.user-service Topic: track.click.v2 Schema Registry Avro Schema 管理 Flink 流处理 实时聚合 / 预警 Spark 批处理 T+1 全量分析 OLAP 引擎 ClickHouse / Doris 所有数据源统一经过 Kafka，Schema 由 Schema Registry 集中管理，下游流批引擎按需消费 数据源与计算层完全解耦，新增数据源只需开发采集端，不影响下游

总结一下

这一章我们聊了：

  • 为什么用 Kafka 做统一数据管道——解耦、缓冲、回溯、多订阅
  • Topic 设计的三条原则——命名规范、分区数估算、数据隔离
  • Avro + Schema Registry 的价值——体积小、有约束、支持演化
  • 三种数据源的接入实战——binlog、日志、埋点

数据源统一接入这块，说白了就是「先标准化，再谈处理」。标准没定好，后面流批一体做得再好，数据质量也上不去。我见过太多项目，前期图省事，后面花几倍的时间来填坑。

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