Apache Avro 数据序列化格式详解

发布时间： 更新时间： 总字数：2228 阅读时间：5m 作者：IP：上海 网址

本文系统介绍 Apache Avro 二进制数据序列化系统：它以 JSON 定义 Schema、以紧凑二进制编码数据、文件头部内嵌 Schema 并支持向后兼容的 Schema 演进，是 Kafka 消息流与数据湖 Raw 层落地的事实标准之一。文章通过 fastavro 实测代码演示 Avro Data File 读写与 schemaless 内存序列化两种模式，并给出与 Parquet、Protobuf、JSON 的选型对比及生产环境常见踩坑点。

Avro 是什么

Apache Avro 是由 Apache Hadoop 项目衍生出的一个开源二进制数据序列化系统。它结合了轻量级的二进制传输格式与强大的模式定义（Schema），广泛用于大数据存储与流式消息传输系统（如 Apache Kafka、Spark、Hadoop）。

核心特性

  1. 强 Schema 约束（基于 JSON 定义）：数据的结构（字段名、数据类型等）使用人类可读的 JSON 格式定义；
  2. 高效的二进制编码：写入和传输时使用紧凑的二进制格式，比传统的 JSON 或 XML 体积小得多，且序列化/反序列化速度极快；
  3. 自描述性（Self-Describing）：存储为磁盘文件（Avro Data File / .avro）时，Schema 会直接保存在文件头部，读取该文件的程序无需提前获取 Schema 即可解析数据；
  4. 优秀的 Schema 演进（Schema Evolution）：支持字段的增加、删除与重命名，读取方和写入方可以使用不同版本的 Schema，只要二者相互兼容；
  5. 行式存储（Row-Oriented）：与列式存储（如 Parquet、ORC）不同，Avro 按行组织数据，写入性能优异，非常适合作为流处理（Kafka Event Stream）或数据摄入（Landing Zone / Staging）的格式。

与其他格式的对比

格式存储方式Schema 支持优势场景
Avro行式（Row）JSON 定义，支持强演进Kafka 消息流、写入密集型、原始数据落地
Parquet列式（Columnar）文件自包含元数据OLAP 分析、查询聚合、Spark/Hive 大规模读取
Protobuf二进制流.proto 文件编译微服务 RPC 通信、实时服务调用
JSON文本/行式可选（JSON Schema）Web API 交互、人类可读性要求高

Python 实战（fastavro）

在 Python 中，处理 Avro 有两个常用库：

  • avro：官方库，实现完整但性能较慢；
  • fastavro（推荐）：采用 C/Cython 优化，读写性能比官方库快数倍至数十倍，是工业界主流。

本文示例均在 fastavro 1.13.1 下实测通过。依赖安装（建议在 python3 -m venv 虚拟环境中执行）：

bash
pip install fastavro

示例一：Avro Data File 写入与读取

python
import fastavro

# 1. 定义 Schema（JSON 格式描述）
schema = {
    "doc": "用户行为日志 Schema",
    "name": "UserEvent",
    "namespace": "com.example.avro",
    "type": "record",
    "fields": [
        {"name": "user_id", "type": "long"},
        {"name": "username", "type": "string"},
        {"name": "email", "type": ["null", "string"], "default": None},  # 可空字段
        {"name": "timestamp", "type": "long"},
        {
            "name": "tags",
            "type": {"type": "array", "items": "string"}  # 列表类型
        }
    ]
}

# 2. 准备数据（行式记录，列表内为字典）
records = [
    {
        "user_id": 1001,
        "username": "Alice",
        "email": "alice@example.com",
        "timestamp": 1710000000,
        "tags": ["vip", "mobile"]
    },
    {
        "user_id": 1002,
        "username": "Bob",
        "email": None,
        "timestamp": 1710000100,
        "tags": ["new_user"]
    }
]

# 3. 写入 Avro 文件 (.avro)
# fastavro.writer 会把 schema 写入文件头部
filename = "users.avro"
with open(filename, "wb") as out_file:
    fastavro.writer(out_file, schema, records)

print(f"数据已成功写入到 {filename}")

# 4. 从 Avro 文件读取数据
print("\n正在读取 Avro 文件内容：")
with open(filename, "rb") as in_file:
    # fastavro.reader 会自动从文件头解析 schema
    avro_reader = fastavro.reader(in_file)

    # 打印内嵌的 schema（全限定名含 namespace 前缀）
    print(f"文件内嵌 Schema 名称: {avro_reader.writer_schema['name']}")

    # 遍历每条记录
    for record in avro_reader:
        print(record)

实测输出：

text
数据已成功写入到 users.avro
文件内嵌 Schema 名称: com.example.avro.UserEvent
{'user_id': 1001, 'username': 'Alice', 'email': 'alice@example.com', 'timestamp': 1710000000, 'tags': ['vip', 'mobile']}
{'user_id': 1002, 'username': 'Bob', 'email': None, 'timestamp': 1710000100, 'tags': ['new_user']}

示例二：内存序列化（Kafka 等网络传输场景）

在 Kafka 等消息队列中，通常不需要保存完整的文件头，而是仅将单个消息对象序列化为 bytes 传输（配合 Confluent Schema Registry 使用）：

python
import io

import fastavro

# 1. 明确定义 Schema
raw_schema = {
    "doc": "用户事件",
    "name": "UserEvent",
    "namespace": "com.example.avro",
    "type": "record",
    "fields": [
        {"name": "user_id", "type": "long"},
        {"name": "username", "type": "string"},
        {"name": "email", "type": ["null", "string"], "default": None},
        {"name": "timestamp", "type": "long"}
    ]
}

# 必须调用 parse_schema 转换为 fastavro 内部优化的解析对象
parsed_schema = fastavro.parse_schema(raw_schema)

# 2. 待序列化数据
record = {
    "user_id": 1001,
    "username": "Alice",
    "email": "alice@example.com",
    "timestamp": 1710000000
}

# 3. 序列化为 bytes（模拟 Producer 发送）
bytes_io = io.BytesIO()
fastavro.schemaless_writer(bytes_io, parsed_schema, record)
avro_bytes = bytes_io.getvalue()

print(f"序列化成功，字节长度: {len(avro_bytes)} 字节")
print(f"二进制内容(16进制): {avro_bytes.hex()}")

# 4. 反序列化 bytes（模拟 Consumer 接收）
# 注意：schemaless 格式不含文件头，反序列化时必须由下游显式提供对应版本的 schema
read_io = io.BytesIO(avro_bytes)
decoded_record = fastavro.schemaless_reader(read_io, parsed_schema)

print(f"反序列化得到的数据: {decoded_record}")

选型建议

  • 什么时候选 Avro？
    • 你在使用 Apache Kafka（结合 Confluent Schema Registry 是行业事实标准）；
    • 需要从多种数据源频繁追加写入到数据湖的 Stage/Raw 层；
    • 数据结构变化频繁，且强依赖 Schema Evolution 保证向后兼容性。
  • 什么时候不建议选 Avro？
    • 做海量数据的 OLAP 聚合分析（如 SELECT AVG(age) FROM users），此时应使用列式的 Parquet。

常见踩坑点

  1. schemaless 格式不携带 Schema：schemaless_writer/schemaless_reader 传输的裸 bytes 不含文件头，Consumer 端必须显式拿到与 Producer 兼容版本的 Schema 才能解码——这正是 Kafka 生态必须搭配 Schema Registry 的根本原因，直接裸传 Schema 变更后的消息会导致解码失败。
  2. Schema 解析对象应跨消息复用：fastavro.parse_schema 有解析开销，Producer 高吞吐场景下应解析一次后复用 parsed_schema，不要每条消息都重新解析原始 JSON Schema。
  3. 新增字段必须带默认值：Avro 的向后兼容（Backward Compatibility）要求数据读取方（新 Schema）能读旧数据（旧 Schema）——新 Schema 中新增的字段必须声明 default，否则兼容性检查（如 Schema Registry 的 BACKWARD 模式）会直接拒绝注册。
  4. Union 类型的 default 必须匹配第一个分支：例如 ["null", "string"] 的默认值只能是 null；把 default 写成 "" 会导致 Schema 非法。这也是可空字段的标准写法是把 "null" 放在 Union 首位的原因。
  5. 字段重命名要用 aliases：直接修改字段名在 Avro 看来是“删字段 + 加字段”，读取旧数据时新字段取不到值；正确做法是新 Schema 保留新名称并通过 aliases 声明旧名。

参考

  1. https://avro.apache.org/docs/
  2. https://fastavro.readthedocs.io/en/latest/
最新评论