Schema Registry'yi Kafka ile Tümleştirme
Veri şemalarını otomatik olarak yönetmek ve zorunlu kılmak için Kafka uygulamalarınızda Schema Registry uygulayın.
Schema Registry'yi Kafka ile Tümleştirme, CoddyKit'te ücretsiz bir Apache Kafka & Stream Processing Fundamentals dersidir. Bu, 4 dersinin 3. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, Apache Kafka & Stream Processing Fundamentals öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. Apache Kafka & Stream Processing Fundamentals kursu toplamda 4 dersten oluşur.
Bu dersin bazı bölümleri henüz çevrilmemiş olup İngilizce olarak gösterilmektedir.
Why Integrate Schema Registry?
You've learned about Kafka and Schema Registry. Now, let's connect them! Integrating Schema Registry into your Kafka applications is vital for ensuring data quality and compatibility.
It acts as a central repository for schemas, allowing producers and consumers to validate and evolve data formats safely.
How it Works: Serializers
To integrate, Kafka clients use special serializers and deserializers that communicate with the Schema Registry.
When a producer sends data, the KafkaAvroSerializer (or Protobuf/JSON Schema equivalent) takes your data, registers its schema (if new), and then prefixes the data with a schema ID before sending it to Kafka.
Producer Configuration Essentials
To make your Kafka producer work with Schema Registry, you need to set specific properties. These tell the producer where the Schema Registry is and which serializer to use.
key.serializer: OftenStringSerializerorKafkaAvroSerializer.value.serializer: Set this toio.confluent.kafka.serializers.KafkaAvroSerializer.schema.registry.url: The URL of your Schema Registry instance (e.g.,http://localhost:8081).
Producer Code: Defining an Avro Schema
Before we send data, we need to define its structure using an Avro schema. For simplicity, we'll create a basic 'User' schema with a name and age field.
This schema will be used to create a GenericRecord.
import org.apache.avro.Schema;
public class AvroSchemaDef {
public static final String USER_SCHEMA_JSON =
"{\"namespace\": \"com.coddykit\", " +
"\"type\": \"record\", " +
"\"name\": \"User\", " +
"\"fields\": [" +
"{\"name\": \"name\", \"type\": \"string\"}," +
"{\"name\": \"age\", \"type\": \"int\"}]}";
public static final Schema USER_SCHEMA =
new Schema.Parser().parse(USER_SCHEMA_JSON);
public static void main(String[] args) {
System.out.println("User Schema Defined!");
}
}Producer Code: Sending Avro Data
Here's a complete Java producer application. Notice how we configure the serializers and the Schema Registry URL. We then create a GenericRecord based on our USER_SCHEMA and send it.
Try running this example!
import org.apache.kafka.clients.producer.*;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import java.util.Properties;
public class AvroProducer {
public static final String USER_SCHEMA_JSON =
"{\"namespace\": \"com.coddykit\", " +
"\"type\": \"record\", " +
"\"name\": \"User\", " +
"\"fields\": [" +
"{\"name\": \"name\", \"type\": \"string\"}," +
"{\"name\": \"age\", \"type\": \"int\"}]}";
public static final Schema USER_SCHEMA =
new Schema.Parser().parse(USER_SCHEMA_JSON);
public static void main(String[] args) {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
props.put("schema.registry.url", "http://localhost:8081");
Producer<String, GenericRecord> producer = new KafkaProducer<>(props);
String topic = "avro-users";
GenericRecord user = new GenericData.Record(USER_SCHEMA);
user.put("name", "Coddy");
user.put("age", 5);
ProducerRecord<String, GenericRecord> record = new ProducerRecord<>(topic, "user-1", user);
try {
producer.send(record, (metadata, exception) -> {
if (exception == null) {
System.out.println("Sent record to topic " + metadata.topic() + " partition " + metadata.partition() + " offset " + metadata.offset());
} else {
exception.printStackTrace();
}
});
} finally {
producer.flush();
producer.close();
}
}
}How it Works: Deserializers
On the consumer side, the KafkaAvroDeserializer (or equivalent) plays the opposite role.
When a consumer receives a message, the deserializer extracts the schema ID, fetches the corresponding schema from Schema Registry, and then uses that schema to correctly deserialize the message back into your application's data type (e.g., a GenericRecord or a specific Avro object).
Consumer Configuration Essentials
Similar to producers, Kafka consumers also need specific properties to work with Schema Registry:
key.deserializer: OftenStringDeserializerorKafkaAvroDeserializer.value.deserializer: Set this toio.confluent.kafka.serializers.KafkaAvroDeserializer.schema.registry.url: The URL of your Schema Registry instance.group.id: A unique ID for your consumer group.auto.offset.reset: Defines behavior when no initial offset is found (e.g.,earliestorlatest).
Consumer Code: Receiving Avro Data
This consumer application is configured to read Avro messages from the 'avro-users' topic. It uses KafkaAvroDeserializer to automatically handle schema resolution.
Run this example AFTER running the producer to see the data!
import org.apache.kafka.clients.consumer.*;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.avro.generic.GenericRecord;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class AvroConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "avro-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
props.put("schema.registry.url", "http://localhost:8081");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
Consumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
String topic = "avro-users";
consumer.subscribe(Collections.singletonList(topic));
System.out.println("Listening for messages on topic: " + topic);
try {
while (true) {
ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, GenericRecord> record : records) {
System.out.printf("Received record (key=%s, value=%s, partition=%d, offset=%d)\n",
record.key(), record.value(), record.partition(), record.offset());
GenericRecord user = record.value();
System.out.println(" User Name: " + user.get("name") + ", Age: " + user.get("age"));
}
}
} finally {
consumer.close();
}
}
}Benefits of Seamless Integration
Integrating Schema Registry with your Kafka clients offers significant advantages:
- Data Compatibility: Ensures producers and consumers always understand each other's data formats.
- Schema Evolution: Safely update schemas over time without breaking existing applications.
- Data Governance: Centralized schema management provides a single source of truth for your data structures.
- Reduced Boilerplate: Serializers/deserializers handle schema management automatically.
Quick Check: Schema Registry Setup
Which of the following properties are essential for a Kafka client (producer or consumer) to integrate with Confluent Schema Registry using Avro?
Recap: Integrating Schema Registry
In this lesson, you learned how to integrate Confluent Schema Registry with your Kafka applications.
- We configured Kafka producers and consumers with `schema.registry.url`.
- We used `KafkaAvroSerializer` and `KafkaAvroDeserializer` to handle Avro data automatically.
- You saw practical examples of sending and receiving `GenericRecord`s.
This integration is key for robust, schema-driven data pipelines!
Sıkça Sorulan Sorular
“Schema Registry'yi Kafka ile Tümleştirme” dersi ücretsiz mi?
Evet — “Schema Registry'yi Kafka ile Tümleştirme” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve Apache Kafka & Stream Processing Fundamentals kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. Apache Kafka & Stream Processing Fundamentals kursu toplamda 4 dersten oluşur.
“Schema Registry'yi Kafka ile Tümleştirme” dersinde ne öğreneceğim?
Veri şemalarını otomatik olarak yönetmek ve zorunlu kılmak için Kafka uygulamalarınızda Schema Registry uygulayın. Apache Kafka & Stream Processing Fundamentals ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.
Apache Kafka & Stream Processing Fundamentals öğrenmeye başlamak için deneyim gerekli mi?
Önceden deneyim gerekmez. CoddyKit'te Apache Kafka & Stream Processing Fundamentals, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 3. dersidir.
“Schema Registry'yi Kafka ile Tümleştirme” dersi ne kadar sürer?
Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.
Bu Apache Kafka & Stream Processing Fundamentals dersinde kod yazıp çalıştırabilir miyim?
Evet. Her Apache Kafka & Stream Processing Fundamentals dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.
Bu kursun tüm dersleri
- Şema Yönetimi Neden Önemlidir?
- Avro ve Protobuf Şemaları
- Schema Registry'yi Kafka ile Tümleştirme
- Şema Evrimi ve Uyumluluk Kipleri