Apache Kafka & Stream Processing Fundamentals · レッスン

Schema RegistryとKafkaの統合

KafkaアプリケーションにSchema Registryを実装し、データスキーマを自動的に管理・適用します。

レッスン 3/411 ステップ

「Schema RegistryとKafkaの統合」はCoddyKit上の無料Apache Kafka & Stream Processing Fundamentalsレッスンです。これはレッスン3/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応の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時間対応のAIチューター）、Apache Kafka & Stream Processing Fundamentalsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Apache Kafka & Stream Processing Fundamentalsコースには全4レッスンが含まれています。

「Schema RegistryとKafkaの統合」で何を学びますか？

KafkaアプリケーションにSchema Registryを実装し、データスキーマを自動的に管理・適用します。ブラウザで直接実行するハンズオンコードでApache Kafka & Stream Processing Fundamentalsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

Apache Kafka & Stream Processing Fundamentalsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのApache Kafka & Stream Processing Fundamentalsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。これはレッスン3/4です。

「Schema RegistryとKafkaの統合」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このApache Kafka & Stream Processing Fundamentalsレッスンでコードを書いて実行できますか？

はい。すべてのApache Kafka & Stream Processing Fundamentalsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

← Apache Kafka & Stream Processing Fundamentalsに戻る