如何让Flink从Avro Schema自动推断所有字段创建表？

当然可以让Flink自动从Avro Schema推断表字段，无需手动逐个指定，以下是两种常用实现方案：

1. 从Schema Registry自动拉取Schema #

如果你的Avro Schema已注册到Schema Registry（如Confluent Schema Registry），只需在WITH参数中配置Registry地址，Flink会自动拉取对应主题的Schema并生成表结构，建表时无需显式定义字段列表。

示例建表语句：

CREATE TABLE user_behavior
WITH (
 'connector' = 'kafka',
 'topic' = 'user_behavior',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'avro',
 'avro.schema-registry.url' = 'http://your-schema-registry:8081'
)

2. 本地指定Avro Schema文件/内容 #

如果Schema未注册到Registry，可直接在配置中指定本地Schema文件路径，或直接写入Schema内容，Flink会基于该Schema推断表字段。

方式一：指定本地Schema文件路径 #

CREATE TABLE user_behavior
WITH (
 'connector' = 'kafka',
 'topic' = 'user_behavior',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'avro',
 'avro.schema-path' = '/path/to/your/user_behavior.avsc'
)

方式二：直接写入Schema内容 #

CREATE TABLE user_behavior
WITH (
 'connector' = 'kafka',
 'topic' = 'user_behavior',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'avro',
 'avro.schema' = '{
   "type": "record",
   "name": "UserBehavior",
   "fields": [
     {"name": "user_id", "type": "long"},
     {"name": "item_id", "type": "long"},
     {"name": "category_id", "type": "long"},
     {"name": "behavior", "type": "string"},
     {"name": "ts", "type": {"type": "long", "logicalType": "timestamp-millis"}}
   ]
 }'
)

注意：使用上述方式时，建表语句中不需要编写字段定义部分，Flink会自动根据指定的Schema生成包含所有字段的表结构，适用于Schema包含数百个字段的场景。

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