...
Code Block | ||||
---|---|---|---|---|
| ||||
StreamTableEnvironment tEnv String schemaRegistryURL = ...; ConfluentSchemaRegistryCatalog Map<String, String> kafkaProps = ...; SchemaRegistryCatalog catalog = new ConfluentSchemaRegistryCatalog( properties, schemaRegistryURL, "catalog1", "db1"); SchemaRegistryCatalog.builder() .schemaRegistryURL(schemaRegistryURL) .kafkaOptions(kafkaProps) .catalogName("myCatalog") .dbName("myDB") .build(); tEnv.registerCatalog("catalog1", catalog); // ---------- Consume stream from Kafka ------------------- String // Assumes there is a topic named 'transactions' String query = "SELECT\n" + " id, amount\n" + "FROM catalog1myCatalog.db1myDB.transactions1";transactions"; |
Introduction to Confluent Schema Registry
Terminology
What is a topic versus a schema versus a subject?
- Subject: Schema Registry defines a scope in which schemas can evolve, and that scope is the subject. The name of the subject depends on the configured subject name strategy, which by default is set to derive subject name from topic name
- Topic:A Kafka topic contains messages, and each message is a key-value pair. Either the message key or the message value, or both, can be serialized as Avro, JSON, or Protobuf
- Schema name:Schema name, for Avro it is the record name, for Json, it is the title name
See terminology-review for details.
Subject Naming Strategy
There are 3 kinds of naming strategy for current 5.5.1 version:
TopicNameStrategy
: <topic name>-key | <topic name>-valueRecordNameStrategy
: <fully-qualified record name>-key | <fully-qualified record name>-valueTopicRecordNameStrategy
: <topic name>-<fully-qualified record name>-key | <topic name>-<fully-qualified record name>-value
The RecordNameStrategy
allows different schemas(record type) within one Kafka topic,the schema compatibility check for same record name are among all the topics,while TopicRecordNameStrategy
checks the compatibility of same record name schema within one Kafka topic.
See sr-schemas-subject-name-strategy for details.
The Restful API
See schemaregistry-api for details.
Design Proposal
ConfluentSchemaRegistryCatalog
...