能否让KSQL直接使用Schema中的默认值填充流数据？

我希望KSQL流能够自动使用Schema中指定的默认值进行填充，除了手动用coalesce语句指定外，是否有其他方法？

我按以下步骤进行了操作：

1. 创建带Schema的topic t1-a

kafka-avro-console-producer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081 --topic t1-a \
--property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"name","type":"string","default":"no-name"}]}'

2. 通过Schema Registry REST API将兼容性设置为FULL
3. 用CLI生成记录：

{"name":"john"}
{"name":"doe"}

4. 更新Schema：

kafka-avro-console-producer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081 --topic t1-a \
--property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"name","type":"string", "default":"no-name"}, {"name":"age","type":"string", "default":"ageless-wonder"}]}'

5. 生成新记录：

{"name":"jack", "age":"100"}
{"name":"jill", "age":"101"}

6. 启动ksql cli并创建流：

CREATE STREAM t1_a WITH (KAFKA_TOPIC='t1-a',VALUE_FORMAT='AVRO');

7. 查询记录：

SELECT * FROM t1_a;

查询结果中John和Doe的Age字段为null，而非Schema指定的默认值"ageless-wonder"。我知道可以用coalesce设置默认值，但能否直接利用已有的Schema默认值填充？

目前KSQL本身不支持直接读取Avro Schema中的默认值来自动填充缺失字段，原因是KSQL处理Avro数据时依赖的KafkaAvroDeserializer默认不会主动应用Schema默认值——仅部分客户端会在字段完全缺失且Schema定义默认值时填充，但KSQL的处理逻辑未触发该行为。

不想手动写coalesce的话，可尝试两种替代方案：

• 在数据生成端填充默认值：
  使用Avro生成类（如通过Avro Tools生成Java类）构造消息，确保生产者发送数据时自动应用Schema默认值，避免发送缺失字段的记录。
• 创建流时定义默认值：
  在KSQL流的DDL中直接指定字段默认值，无需依赖Schema Registry中的配置。示例：

  CREATE STREAM t1_a (
    name STRING DEFAULT 'no-name',
    age STRING DEFAULT 'ageless-wonder'
  ) WITH (KAFKA_TOPIC='t1-a', VALUE_FORMAT='AVRO');
  

  这种方式虽需在DDL中重复默认值，但能让KSQL自动为缺失字段填充指定值，无需每次查询都写coalesce。

需注意，这两种方案都无法完全“直接复用Schema Registry中已定义的默认值”，因为KSQL当前设计并未同步Schema默认值到流元数据中。若需完全依赖Schema默认值，可能需等待KSQL版本更新，或自定义UDF读取Schema Registry中的默认值并应用，但后者实现成本较高。

内容的提问来源于Stack Exchange，提问作者freddie-mercurial