1. 课程导入与全景认知

在开始学习 Confluent Kafka 数据质量与元数据治理之前，先要建立一个清晰的全景认知：这门课不是单纯讲 Kafka 的收发消息，也不是只讲 Schema Registry 的 API 用法，而是围绕“数据在流式管道中如何被定义、约束、观测、追溯和治理”展开。Kafka 本身负责高吞吐、可持久化的事件传输，但它并不天然保证消息内容正确、字段含义一致、上下游契约稳定。Confluent 平台在 Kafka 之上补充了 Schema Registry、Stream Governance、Data Contract、流式血缘、质量规则与元数据目录等能力，使数据质量与元数据治理从“事后补救”前移到“生产端约束”和“消费端可观测”。

本课程讨论的边界可以概括为三层：第一层是数据契约层，包括 Avro、Protobuf、JSON Schema 的注册、兼容性策略、版本演进和 Schema 引用；第二层是数据质量层，包括字段级校验、空值率、枚举越界、数值范围、消息重复与延迟等质量规则的表达与监控；第三层是元数据治理层，包括 Topic、Schema、Consumer Group、流式血缘、标签、所有者、SLA 和审计信息。三层不是孤立工具，而是通过 Confluent Cloud 或自建 Confluent Platform 的控制面与数据面协同工作。下图给出本课程的整体能力地图与学习路径，后续章节会沿着这条主线逐步展开。

从图中可以看到，学习路径不是从 Kafka 基础命令开始，而是从“契约”切入。原因很直接：如果生产端写入的消息没有 Schema 约束，消费端拿到的就是一堆无法解释的字节；如果 Schema 演进没有兼容性策略，一次字段删除就可能让下游作业全部失败。因此，本课程会先建立数据契约，再叠加质量规则，最后用元数据把整条链路串起来。

核心挑战集中在四个方面。第一，Schema 演进与兼容性：Avro 的 BACKWARD、FORWARD、FULL 策略分别对应不同的消费/生产升级顺序，选错策略会导致新旧客户端无法共存。第二，质量规则的表达位置：规则放在生产端拦截，还是放在消费端观测，直接影响延迟和可恢复性。第三，元数据的自动采集与人工标注：流式血缘需要从 Consumer Group 的 offset 提交、连接器配置和 SQL 作业中推断，但业务含义、所有者、SLA 必须人工补充。第四，治理与性能的平衡：每条消息都做复杂校验会拖高延迟，需要区分同步拦截和异步采样。

为了让后续章节有统一的实验对象，这里先约定一套贯穿课程的示例场景：一个电商订单流，Topic 名为 orders.raw.v1，消息体包含 order_id、user_id、amount、currency、status、event_time 六个字段。其中 order_id 为字符串主键，amount 为 decimal 金额，currency 限定为 CNY/USD/EUR，status 限定为 CREATED/PAID/SHIPPED/CANCELLED。后续的 Schema 注册、兼容性测试、质量规则和血缘分析都会围绕这个 Topic 展开。示例参数如下，均为教学假设值，不是实测结果：

Topic: orders.raw.v1
Partitions: 6
Replication Factor: 3
Schema Format: Avro
Compatibility: BACKWARD
Quality Rules:
  - order_id NOT NULL, 长度 16~32
  - amount > 0 且 <= 100000
  - currency IN (CNY, USD, EUR)
  - status IN (CREATED, PAID, SHIPPED, CANCELLED)
  - event_time 与当前时间偏差 <= 300s

实战环境准备分两种路线。如果使用 Confluent Cloud，需要创建 Environment、Cluster、API Key，并在 Schema Registry 中开启 Stream Governance Essentials；如果使用本地 Docker，可以用 confluent-local 或 cp-all-in-one 启动 Broker、Schema Registry、Control Center 和 ksqlDB。无论哪条路线，都要准备三个客户端工具：confluent CLI 用于管理 Topic 和 Schema，kafka-avro-console-producer 用于写入带 Schema 的消息，kafka-avro-console-consumer 用于验证消费端反序列化。下面给出本地环境的最小验证命令，用于确认 Schema Registry 与 Broker 已连通：

# 1. 检查 Schema Registry 是否可用
curl -s http://localhost:8081/subjects | jq .

# 2. 创建示例 Topic
confluent kafka topic create orders.raw.v1 --partitions 6 --config retention.ms=604800000

# 3. 注册初始 Avro Schema
curl -X POST http://localhost:8081/subjects/orders.raw.v1-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"user_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"event_time\",\"type\":\"long\"}]}"}'

# 4. 写入一条测试消息
echo '{"order_id":"ORD20250101000001","user_id":"U10086","amount":199.00,"currency":"CNY","status":"CREATED","event_time":1735689600000}' | \
  kafka-avro-console-producer --broker-list localhost:9092 --topic orders.raw.v1 \
  --property schema.registry.url=http://localhost:8081 \
  --property value.schema.id=1