本文档是 Apache Flink 的旧版本。建议访问 最新的稳定版本。
  1. Connectors
  2. Table & SQL Connectors
  3. Formats
  4. Confluent Avro

Confluent Avro Format

Format: Serialization Schema Format: Deserialization Schema

Avro Schema Registry (avro-confluent) 格式能让你读取被 io.confluent.kafka.serializers.KafkaAvroSerializer序列化的记录，以及可以写入成能被 io.confluent.kafka.serializers.KafkaAvroDeserializer反序列化的记录。

当以这种格式读取（反序列化）记录时，将根据记录中编码的 schema 版本 id 从配置的 Confluent Schema Registry 中获取 Avro writer schema ，而从 table schema 中推断出 reader schema。

当以这种格式写入（序列化）记录时，Avro schema 是从 table schema 中推断出来的，并会用来检索要与数据一起编码的 schema id。我们会在配置的 Confluent Schema Registry 中配置的 subject 下，检索 schema id。subject 通过 avro-confluent.schema-registry.subject 参数来制定。

Avro Schema Registry 格式只能与Apache Kafka SQL连接器结合使用。

依赖

In order to use the Avro Schema Registry format the following dependencies are required for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.

Maven dependency SQL Client JAR
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-avro-confluent-registry</artifactId>
  <version>1.12.7</version>
</dependency> 		Download

如何创建使用 Avro-Confluent 格式的表

以下是一个使用 Kafka 连接器和 Confluent Avro 格式创建表的示例。

CREATE TABLE user_behavior (
  user_id BIGINT,
  item_id BIGINT,
  category_id BIGINT,
  behavior STRING,
  ts TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = 'localhost:9092',
  'topic' = 'user_behavior'
  'format' = 'avro-confluent',
  'avro-confluent.schema-registry.url' = 'http://localhost:8081',
  'avro-confluent.schema-registry.subject' = 'user_behavior'
)

Format 参数

参数 是否必选 默认值 类型 描述
format
 required (none) String Specify what format to use, here should be 'avro-confluent'.
avro-confluent.basic-auth.credentials-source
 optional (none) String Basic auth credentials source for Schema Registry
avro-confluent.basic-auth.user-info
 optional (none) String Basic auth user info for schema registry
avro-confluent.bearer-auth.credentials-source
 optional (none) String Bearer auth credentials source for Schema Registry
avro-confluent.bearer-auth.token
 optional (none) String Bearer auth token for Schema Registry
avro-confluent.ssl.keystore.location
 optional (none) String Location / File of SSL keystore
avro-confluent.ssl.keystore.password
 optional (none) String Password for SSL keystore
avro-confluent.ssl.truststore.location
 optional (none) String Location / File of SSL truststore
avro-confluent.ssl.truststore.password
 optional (none) String Password for SSL truststore
avro-confluent.schema-registry.subject
 optional (none) String The Confluent Schema Registry subject under which to register the schema used by this format during serialization. By default, 'kafka' and 'upsert-kafka' connectors use '<topic_name>-value' or '<topic_name>-key' as the default subject name if this format is used as the value or key format. But for other connectors (e.g. 'filesystem'), the subject option is required when used as sink.
avro-confluent.schema-registry.url
 required (none) String The URL of the Confluent Schema Registry to fetch/register schemas.

数据类型映射

目前 Apache Flink 都是从 table schema 去推断反序列化期间的 Avro reader schema 和序列化期间的 Avro writer schema。显式地定义 Avro schema 暂不支持。 Apache Avro Format中描述了 Flink 数据类型和 Avro 类型的对应关系。

除了此处列出的类型之外，Flink 还支持读取/写入可为空（nullable）的类型。 Flink 将可为空的类型映射到 Avro union(something, null), 其中 something 是从 Flink 类型转换的 Avro 类型。

您可以参考 Avro Specification 以获取有关 Avro 类型的更多信息。

想参与贡献翻译？