4、ODS层实战：基于Kafka的实时数据接入、JSON/AVRO格式处理

ODS层，说白了就是实时数仓的“第一道门”。

所有业务系统的数据，不管你是MySQL的Binlog、APP的埋点日志，还是IoT设备的传感器数据，都得先经过ODS层。我习惯把它比作“数据码头”——船（数据）先靠岸，再决定往哪个仓库（后续分层）送。

这一章，我们重点解决两个问题：怎么把数据实时接进来，以及怎么处理JSON和AVRO这两种主流格式。

4.1 为什么选Kafka作为ODS层核心载体？

你可能会问：为什么非要用Kafka？用消息队列不行吗？

嗯，这里要注意。Kafka在实时数仓里的角色，不只是消息队列那么简单。它更像一个“分布式日志存储系统”。我曾在项目中见过有人用RabbitMQ做ODS层，结果数据量一上来，消费端跟不上，消息积压直接把内存撑爆了。

Kafka的优势，我总结了三句话：

• 高吞吐：单机每秒几十万条消息，轻松应对业务高峰。
• 持久化：数据落盘，可以回溯重放。你想想看，如果下游挂了，数据还能找回来，这多重要。
• 多订阅者：一份数据，多个消费者各取所需。比如实时ETL用一份，离线批处理用另一份。

4.2 实时数据接入：从源头到Kafka

数据接入的方式，取决于你的数据源。我挑三个最常见的场景来讲。

4.2.1 MySQL Binlog 接入（Canal + Kafka）

这是实时数仓最经典的接入方式。业务库的增删改，通过Canal解析Binlog，再推送到Kafka。

我曾经踩过一个坑：Canal默认只推送增量数据，但如果你要初始化全量数据，就得手动做一次“全量+增量”的衔接。后来我写了个脚本，先拉全量到Kafka的一个独立Topic，再切到增量Topic，下游通过时间戳做去重。

配置示例（Canal的instance.properties）：

# 指定要监听的数据库和表
canal.instance.filter.regex=test_db\\..*
# 投递到Kafka的Topic
canal.mq.topic=ods_binlog_test_db
# 分区数，建议和Kafka分区数一致
canal.mq.partitions=6

4.2.2 日志文件接入（Filebeat + Kafka）

APP埋点、服务器日志，通常走Filebeat或Flume。我个人更倾向Filebeat，因为它轻量，资源占用低。

配置示例（Filebeat的filebeat.yml）：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  fields:
    log_type: app_log

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "ods_app_log"
  partition.round_robin:
    reachable_only: true

这里有个小技巧：在fields里加一个log_type标签，下游可以根据这个标签做路由，不用再解析日志内容来判断来源。

4.2.3 自定义API接入（HTTP + Kafka Producer）

有些第三方系统，既没有Binlog，也不产生日志文件，只提供HTTP接口。这时候，你需要自己写一个数据采集服务。

我建议用Spring Boot + Kafka Producer封装一个通用接入层。核心代码就几行：

@RestController
public class DataCollectController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping("/collect")
    public String collect(@RequestBody String data) {
        // 直接投递到Kafka，不做任何处理
        kafkaTemplate.send("ods_raw_data", data);
        return "success";
    }
}

注意：这里一定要用异步发送，否则接口响应时间会拖垮上游。

4.3 JSON格式处理：灵活但“吃”性能

JSON是ODS层最常见的格式。为什么？因为业务系统改字段太频繁了。JSON的schema-less特性，让上下游解耦。

但JSON有个致命问题：解析性能差。我做过压测，同样的数据量，JSON解析比AVRO慢3-5倍。而且JSON是文本格式，占用存储空间大。

JSON处理示例（Flink SQL）：

CREATE TABLE ods_json (
    user_id STRING,
    event_type STRING,
    event_time BIGINT,
    properties MAP<STRING, STRING>
) WITH (
    'connector' = 'kafka',
    'topic' = 'ods_app_log',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true'
);

注意json.ignore-parse-errors这个参数。我建议设为true，因为生产环境里，偶尔会有一条格式异常的JSON。如果直接报错，整个任务就挂了。设为true后，异常数据会被丢弃，保证任务持续运行。

4.4 AVRO格式处理：高效但需要Schema管理

AVRO是二进制格式，解析快、体积小。如果你对性能有极致要求，或者数据量特别大（比如每天几百TB），我强烈推荐用AVRO。

但AVRO有个门槛：必须定义Schema。而且Schema一旦定义，改起来很麻烦。我曾在项目中因为改了AVRO Schema，忘记通知下游，结果下游反序列化直接报错，排查了半天。

4.4.1 AVRO Schema定义

一个典型的AVRO Schema长这样：

{
  "type": "record",
  "name": "UserEvent",
  "namespace": "com.example.ods",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "event_type", "type": "string"},
    {"name": "event_time", "type": "long"},
    {"name": "properties", "type": {"type": "map", "values": "string"}}
  ]
}

4.4.2 Schema Registry：解决版本兼容问题

为了解决Schema变更的问题，业界普遍使用Schema Registry。它就像一个Schema的“Git仓库”，每个版本都有记录。

我建议用Confluent Schema Registry，它支持三种兼容性策略：

策略	说明	适用场景
BACKWARD	新Schema可以读旧数据	新增可选字段
FORWARD	旧Schema可以读新数据	删除字段
FULL	双向兼容	严格约束的场景

4.5 JSON vs AVRO：怎么选？

我直接给结论：

• 选JSON：业务变化快、字段经常增删、团队对AVRO不熟。
• 选AVRO：数据量大（日均百亿条以上）、对性能敏感、有完善的Schema管理流程。

如果你实在拿不准，我建议ODS层用JSON，DWD层转AVRO。这样既保留了ODS的灵活性，又享受了AVRO的高性能。

4.6 本章核心知识体系

下面这张图，是我对ODS层数据接入和格式处理的总结。你可以把它当作一个“操作手册”。

4.7 避坑总结

最后，我把ODS层实战中常见的坑列出来，你遇到了可以直接翻到这一页：

• 数据丢失：Kafka的acks参数一定要设为all，否则Leader挂了会丢数据。
• 重复数据：Canal或Filebeat在网络抖动时会重发，下游必须做幂等处理。
• Schema不兼容：AVRO格式一定要用Schema Registry，并设置BACKWARD策略。
• 性能瓶颈：JSON解析太慢？试试用AVRO，或者用SIMD加速的JSON解析库（比如simdjson）。