Flink能否类似Spark自动推断Kafka topic DDL 无需手动编写CREATE TABLE

Flink 1.17及以上版本已经原生支持Kafka Topic的DDL自动推断能力，无需手动编写完整的CREATE TABLE字段定义，功能和Spark的对应能力对齐，以下是两种常见使用场景：

场景1：消息已在Schema Registry注册 #

如果你的Kafka消息使用Avro/Protobuf等序列化格式，且Schema已经上传到Confluent Schema Registry等服务，仅需要指定核心连接器配置，Flink会自动拉取Schema生成表结构：

CREATE TABLE auto_generated_kafka_table
WITH (
  'connector' = 'kafka',
  'topic' = '替换为你的Topic名称',
  'properties.bootstrap.servers' = '替换为你的Kafka地址:端口',
  'properties.group.id' = '替换为你的消费组ID',
  'format' = 'avro-confluent', -- 可以替换为protobuf-confluent等对接Schema Registry的格式
  'avro-confluent.schema-registry.url' = '替换为你的Schema Registry访问地址'
);

场景2：无Schema Registry的JSON格式消息 #

如果你的Kafka消息为JSON格式且没有注册Schema，可以开启JSON格式自带的Schema推断能力，Flink会拉取指定数量的样本消息自动解析字段和对应数据类型：

CREATE TABLE auto_json_kafka_table
WITH (
  'connector' = 'kafka',
  'topic' = '替换为你的JSON Topic名称',
  'properties.bootstrap.servers' = '替换为你的Kafka地址:端口',
  'properties.group.id' = '替换为你的消费组ID',
  'format' = 'json',
  'json.infer-schema' = 'true', -- 开启自动推断
  'json.infer-schema.samples' = '200' -- 可选配置，指定用来推断的样本数量，默认值为1000
);

补充说明 #

• 如果你使用的Flink版本低于1.17，可以通过自定义Kafka Catalog的方式实现类似能力，批量拉取Topic元数据自动生成表定义，无需逐条编写CREATE TABLE语句。
• 自动推断的字段类型不符合业务预期时，可以手动指定部分字段的名称和类型，其余未显式声明的字段依然会沿用自动推断的结果。

内容的提问来源于stack exchange，提问作者gaurav miglani