Schema Registry catalog
The Schema Registry catalog allows you to access Kafka topics with registered schemas as Flink SQL tables. You can add Schema Registry as a catalog in Flink SQL by adding the dependency to your project, registering it in Java, and enabling it in the custom environment file.
Each Kafka topic will be mapped to a table with TableSchema
that matches the
Avro schema.
Maven Dependency
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-cloudera-registry</artifactId>
<version>1.16.1-csa1.10.0.0</version>
</dependency>
The following example shows how to register and use the Schema Registry catalog from
Java:
SchemaRegistryClient client = new SchemaRegistryClient(
ImmutableMap.of(
SchemaRegistryClient.Configuration.SCHEMA_REGISTRY_URL.name(),
"http://<your_hostname>:7788/api/v1"
)
);
Map<String, String> connectorProps = new Kafka()
.property(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"<your_hostname>:9092")
.startFromEarliest()
.toProperties();
tableEnv.registerCatalog(
"registry", new ClouderaRegistryCatalog("registry", client, connectorProps)
);
tableEnv.useCatalog("registry");