Confluent Kafka主题仅ID字段有值其余为null问题排查

1. 字段名称映射不匹配 #

你的Avro Value Schema中字段名是ID（大写）、name、category等，若Python生产者从MySQL取数据时，构造消息的字典键名和Schema字段名完全不一致（包括大小写），Avro会自动用默认值null填充未匹配的字段。

比如MySQL表字段是id（小写）、product_name，但你构造消息时没将product_name映射到Schema的name字段，就会导致name显示为null。

解决：

• 逐一核对MySQL表字段名、Python代码中消息字典的键名、Avro Schema的字段名，确保三者完全一致（大小写也要严格匹配）。
• 示例映射代码：

# 假设MySQL查询结果的字段为id, product_name, product_category, product_price, update_time
value = {
    "ID": row["id"],
    "name": row["product_name"],
    "category": row["product_category"],
    "price": float(row["product_price"]),
    "last_updated": int(row["update_time"].timestamp() * 1000)
}

2. 数据类型不兼容 #

Avro对数据类型要求严格，不符合Schema定义的类型会被序列化为null：

• price字段Schema定义为["null", "float"]，若MySQL中是decimal类型，直接赋值会导致类型不匹配，需转成float。
• last_updated是timestamp-millis类型，必须传入毫秒级时间戳整数，若直接传datetime对象或字符串，会序列化失败，字段变为null。

解决：

• 强制转换数据类型匹配Schema：
  • 数值类型：将decimal转成float/int（注意精度取舍）。
  • 时间类型：把MySQL的datetime对象转为毫秒级时间戳（如int(datetime_obj.timestamp() * 1000)）。

3. Schema注册与使用不一致 #

确保Python生产者使用的Schema和Confluent Control Center中显示的Schema完全一致：

• 检查代码中硬编码的Schema或从Schema Registry获取的Schema，是否和Control Center里的版本、字段定义一致。
• 若Schema有更新，需保证生产者使用最新版本，且字段新增时已设置默认值（你的Schema已设置，无需担心兼容性问题）。

4. 生产者序列化配置错误 #

确认Avro生产者的序列化器配置正确，避免因配置缺失导致字段无法正常序列化：

from confluent_kafka.avro import AvroProducer

producer_config = {
    'bootstrap.servers': '你的Kafka地址',
    'schema.registry.url': '你的Schema Registry地址',
    'acks': 'all'
}

# 直接使用你的Value和Key Schema字符串
value_schema = """{
  "namespace": "com.kaggle.onlineretail",
  "name": "product",
  "type": "record",
  "fields": [
    {"name": "ID", "type": ["null", "int"], "default": null},
    {"name": "name", "type": ["null", "string"], "default": null},
    {"name": "category", "type": ["null", "string"], "default": null},
    {"name": "price", "type": ["null", "float"], "default": null},
    {"name": "last_updated", "type": ["null",{"type": "long","logicalType": "timestamp-millis"}]}
  ]
}"""

key_schema = """{
    "namespace": "com.kaggle.onlineretail",
    "name": "Product",
    "type": "record",
    "fields": [{"name": "Id", "type": "int"}]
}"""

producer = AvroProducer(producer_config,
                        default_value_schema=value_schema,
                        default_key_schema=key_schema)

快速验证方法 #

在代码中序列化后立即反序列化，检查字段是否正常：

from confluent_kafka.avro import AvroDeserializer
from confluent_kafka.avro.serializer import SchemaRegistryClient
from datetime import datetime

schema_registry_client = SchemaRegistryClient({'url': '你的Schema Registry地址'})
value_deserializer = AvroDeserializer(schema_registry_client=schema_registry_client, schema_str=value_schema)

# 构造测试数据
test_value = {
    "ID": 1,
    "name": "测试商品",
    "category": "测试分类",
    "price": 19.99,
    "last_updated": int(datetime.now().timestamp() * 1000)
}

# 序列化+反序列化
serialized = producer.value_serializer(test_value)
deserialized = value_deserializer(serialized, None)
print(deserialized)  # 输出应包含所有非null字段

内容的提问来源于stack exchange，提问作者Prakhara Kumar