Apache Kafka & Stream Processing Fundamentals · 강의

Schema Registry와 Kafka 통합

Kafka 애플리케이션에 Schema Registry를 구현하여 데이터 스키마를 자동으로 관리하고 적용합니다.

레슨 3/411개 단계

Schema Registry와 Kafka 통합은(는) CoddyKit의 무료 Apache Kafka & Stream Processing Fundamentals 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 Apache Kafka & Stream Processing Fundamentals 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. Apache Kafka & Stream Processing Fundamentals 강의에는 총 4개의 강의가 포함되어 있습니다.

이 강의의 일부는 아직 번역되지 않았으며 영어로 표시됩니다.

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: Often StringSerializer or KafkaAvroSerializer.
value.serializer: Set this to io.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: Often StringDeserializer or KafkaAvroDeserializer.
value.deserializer: Set this to io.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., earliest or latest).

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!

무료로 시작

AI 튜터와 함께 Apache Kafka & Stream Processing Fundamentals을(를) 배우세요 — 무료

브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.

코스: 12
레슨: 48

무료로 학습 시작 전체 코스 보기

자주 묻는 질문

“Schema Registry와 Kafka 통합” 강의는 무료인가요?

네 — “Schema Registry와 Kafka 통합” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 Apache Kafka & Stream Processing Fundamentals 강의 전체를 잠금 해제할 수 있습니다. Apache Kafka & Stream Processing Fundamentals 강의에는 총 4개의 강의가 포함되어 있습니다.

“Schema Registry와 Kafka 통합”에서 뭘 배우나요?

Kafka 애플리케이션에 Schema Registry를 구현하여 데이터 스키마를 자동으로 관리하고 적용합니다. 브라우저에서 직접 실행하는 실습 코드로 Apache Kafka & Stream Processing Fundamentals을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

Apache Kafka & Stream Processing Fundamentals을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 Apache Kafka & Stream Processing Fundamentals은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.

“Schema Registry와 Kafka 통합” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 Apache Kafka & Stream Processing Fundamentals 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 Apache Kafka & Stream Processing Fundamentals 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← Apache Kafka & Stream Processing Fundamentals(으)로 돌아가기