如何阻止Flink SQL自动上传Avro Schema，改用已上传的Schema？

你遇到的问题核心是配置项层级错误：properties.auto.register.schemas是Kafka客户端的配置，而控制Confluent Schema Registry自动注册的是Flink avro-confluent格式的专属配置，需要调整配置位置并补充必要的匹配规则。

解决方案步骤 #

1. 修正自动注册开关配置 #

将原来的properties.auto.register.schemas = 'false'替换为格式层面的配置：

'value.avro-confluent.auto-register-schemas' = 'false'

这个配置才是真正控制Flink是否自动向Schema Registry上传生成的Avro Schema的开关，之前的配置属于Kafka客户端属性，无法影响Flink的Schema注册逻辑。

2. 确保Flink匹配已上传的Schema #

由于你上传的Schema与Flink自动生成的Schema在name、namespace等元信息上不一致，即使关闭自动注册，Flink可能仍无法找到匹配的已有Schema。可以通过以下两种方式明确指定要使用的Schema：

方式一：通过Schema ID指定

找到你上传的Schema在Registry中的ID（可通过Schema Registry API查询），添加配置：

'value.avro-confluent.schema-id' = '你的Schema ID'

方式二：直接指定Schema内容

将你上传的Avro Schema压缩为单行字符串，添加到配置中：

'value.avro-confluent.schema' = '{"type":"record","name":"OutputEvent","namespace":"com.example","doc":"Avro schema for a transformed PoC event","version":1,"fields":[{"name":"category_type","type":"string","doc":"Type of expense category"},{"name":"user_id","type":"string","doc":"UUID uniquely identifying a user"},{"name":"amount","type":"long","doc":"The total amount spent in cents"},{"name":"time_window","type":"long","doc":"The window of this aggregate"}]}'

修改后的完整表定义示例 #

CREATE TABLE aggregate
(
    category_type           STRING NOT NULL,
    user_id                 STRING NOT NULL,
    amount                  BIGINT NOT NULL,
    time_window             BIGINT NOT NULL
) WITH (
    'connector' = 'kafka',
    'topic' = 'output',
    'properties.bootstrap.servers' = 'localhost:9093',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.schema-registry.url' = 'http://localhost:8082',
    'value.avro-confluent.schema-registry.subject' = 'output-value',
    'value.avro-confluent.auto-register-schemas' = 'false',
    'value.avro-confluent.schema-id' = '1' -- 替换为实际的Schema ID
);

注意事项 #

• 确保Flink表的字段名、类型与你上传的Avro Schema完全一致，否则会出现序列化失败。
• 若使用Schema ID方式，需确认该ID对应的Schema在Registry中存在且未被删除。

内容的提问来源于stack exchange，提问作者Jonck van der Kogel