Python × Confluent Schema Registry：Confluent Schema Registry 入門

## 概要

Confluent Schema Registry は、Apache Kafka 上で流通するデータのスキーマ（Avro、JSON Schema、Protobuf など）を一元的に管理し、スキーマ進化時の互換性チェックやバージョン管理を容易にするコンポーネントです。Confluent 社によって開発され、**Confluent Community License** の下で配布されています。これにより、スキーマを独立した「契約」として扱うことで、プロデューサーやコンシューマー間の事前合意や明示的なスキーマ共有の手間を軽減し、システム全体の柔軟性や保守性を向上させます。

また、Python から操作する場合は、`confluent-kafka` ライブラリが提供する `SchemaRegistryClient` クラスを通じて、スキーマの登録・取得・互換性設定・削除など、多彩な機能をプログラムで実行可能です。

本記事の文章とコードの大部分は、 ChatGPT o1 pro mode で記述しています。ライブラリに関するドキュメント内容を提供してサンプルコードの作成からはじめました。それを微修正して、本記事になりました。ChatGPT Pro はおすすめです。

> Schema Registry が学習するために SchemaRegistryClient インスタンスに関して学習するコンテンツを作成して。
> ```text
> {`confluent-kafka` ライブラリの Schema Registry クライアントクラスのドキュメント内容}
> ```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/09ca0a2e-25a9-59f4-846f-aaaff5759646.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7b853a55-f182-c195-e9d4-933a55961dc9.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9e2ce975-ffc5-059a-a539-cc2e26ce2727.png)

本記事は、以下のシリーズの一部です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1ff6386e-0d84-e62b-dd54-8e6164c4ba5b.png)

引用元：[Data in Motion 実現へ：Data Streaming Platform である Confluent の全貌 #Kafka - Qiita](https://qiita.com/manabian/items/f8281514e9afb9a8701c)

## Confluent Schema Registry とは

### Confluent Schema Registry 概要

Confluent Schema Registry は、Apache Kafka 環境におけるスキーマ管理を行うためのコンポーネントです。これにより Kafka メッセージのスキーマ（Avro、JSON Schema、Protobufなど）を集中管理し、スキーマ進化時の互換性担保やバージョン管理を容易にします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/01d5ea32-b57c-5df8-d47d-00b3889bb0bf.png)

引用元：[Schema Registry for Confluent Platform | Confluent Documentation](https://docs.confluent.io/platform/current/schema-registry/index.html)

Confluent Schema Registry は以下のような特徴を持ちます。

- **集中管理**: すべてのスキーマを一元的に管理し、クライアントからのリクエストに応じてスキーマを提供します。  
- **互換性ポリシーの適用**: 新規バージョンのスキーマ登録時に、後方互換性や前方互換性などのポリシーを自動的にチェックし、問題のある変更を防ぎます。  
- **複数のスキーマ形式対応**: Avro、JSON Schema、Protobuf など、異なるスキーマ形式を統一的に扱えるため、異なるアプリケーション間でのスキーマ管理が容易になります。  

Confluent 上では`Data contract`として分類されており、Schema Registry の Subject を確認することできます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e436aa63-994b-a411-e710-3b8a99b08111.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/110a6d98-4dca-7334-f44a-a96164c5fb98.png)


Confluent Community License の下で配布されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c45fc565-cd96-2afa-6b9d-2af34f4c837e.png)

引用元：[confluentinc/schema-registry: Confluent Schema Registry for Kafka](https://github.com/confluentinc/schema-registry?tab=readme-ov-file#license)

### Python で Confluent Schema Registry を操作する方法

Python で Confluent Schema Registry を操作する際には、`confluent-kafka` ライブラリに含まれる Schema Registry クライアントクラス (`SchemaRegistryClient`) を用います。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/13d84170-f886-d44e-4f54-05e188475868.png)

引用元：[confluent_kafka API — confluent-kafka 2.6.0 documentation](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#schemaregistryclient)

このライブラリを使うことで、Python アプリケーションから次のような操作が可能です。

- Schema Registry への接続設定（URL、認証情報、SSL 設定など）  
- スキーマの登録、取得、バージョン管理、削除  
- 互換性レベルの確認・変更、および新スキーマの互換性テスト  
- エラーハンドリング（未登録のスキーマやサブジェクトへのアクセス、互換性違反など）

`SchemaRegistryClient` を利用するためには、Schema Registry の接続情報や認証情報、SSL 設定などを `dict` 形式のコンフィグにまとめて渡します。

| 設定キー | 型  | 説明 |
|-----------|-----|-----|
| `url` *   | str | Schema Registry の URL (必須) |
| `ssl.ca.location` | str | CA 証明書ファイルへのパス |
| `ssl.key.location` | str | クライアント秘密鍵(PEM)へのパス |
| `ssl.certificate.location` | str | クライアント公開鍵(PEM)へのパス |
| `basic.auth.user.info` | str | `username:password` 形式の認証情報 |

`SchemaRegistryClient` はスキーマのライフサイクルを管理するための様々なメソッドを提供しています。主なものを以下に示します。

- スキーマ登録・取得
    - `register_schema(subject_name, schema, normalize_schemas=False)`: サブジェクトにスキーマを登録し、スキーマIDを返します。
    - `get_schema(schema_id)`: スキーマIDに対応するスキーマを取得します。
    - `lookup_schema(subject_name, schema, normalize_schemas=False)`: 指定のスキーマが既に登録済みかを確認します。
- サブジェクト管理
    - `get_subjects()`: 登録されている全サブジェクト一覧を取得します。
    - `delete_subject(subject_name, permanent=False)`: サブジェクトを削除します。(通常は開発・テスト用)
- バージョン管理
    - `get_latest_version(subject_name)`: 最新バージョンのスキーマ情報を取得します。
    - `get_version(subject_name, version)`: 指定バージョンのスキーマ情報を取得します。
    - `get_versions(subject_name)`: サブジェクト下の全バージョン番号を取得します。
    - `delete_version(subject_name, version)`: 指定バージョンを削除します。
- 互換性管理
    - `set_compatibility(subject_name=None, level=None)`: グローバルまたは対象サブジェクトの互換性レベルを設定します。
    - `get_compatibility(subject_name=None)`: 現在の互換性レベルを取得します。
    - `test_compatibility(subject_name, schema, version='latest')`: 提案スキーマが特定バージョンと互換性があるかをチェックします。

### 他社の Schema Registry との差異

Azure や AWS などのクラウドベンダーも、独自の Schema Registry 機能を提供しています。しかし、これらは Confluent Schema Registry とは異なるサービスのようです。

- **Azure Schema Registry**: Azure Event Hubs 環境に最適化されており、Microsoft のエコシステムと緊密に統合されています。  
- **AWS Glue Schema Registry**: AWS 環境（特に Amazon MSK や他のデータストリーミングサービス）でのシームレスなスキーマ管理を目的として Amazon が独自開発しています。

それぞれが GitHub にて公開されています。

- [Azure/azure-schema-registry-for-kafka](https://github.com/Azure/azure-schema-registry-for-kafka)
- [awslabs/aws-glue-schema-registry](https://github.com/awslabs/aws-glue-schema-registry)

## Confluent Schema Registry の実践

### 事前準備

#### Confluent Cloud 上に新規の envrionment を作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/893c1379-86dd-6bba-10be-14e183c138b8.png)

#### Confluent Cloud 上にクラスターを作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a18f1022-3fd2-7f25-bfb6-c8b0f5d4f5c2.png)

#### Schema Registry の認証情報を取得

Envrionment の画面にて右下にある`Stream Governance API`にある`+ Add key`から認証情報を取得します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0b92b0df-118f-0893-7ec3-7f329e6a0451.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fe68c4c8-969e-6b59-6ff8-289fbd5be946.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/50eb637b-7ce9-9019-5d44-d90d449694dd.png)

### Python 上での事前準備

#### confluent-kafka のインストール

Google Colab 上にノートブックを作成して、下記のコードを実行します。

```pyhton
!pip install confluent-kafka -q
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/813fd772-6f36-e588-9f1e-b642a8d80dcd.png)

#### SchemaRegistryClient のインスタンス生成

```python
# Schema Registry の接続情報をセット
sr_url = "<schema.registry.url>"
sr_api_key = "<SR_API_KEY>"
sr_api_secret = "<SR_API_SECRET>"
sr_config = {
    'url': sr_url,
    'basic.auth.user.info': f"{sr_api_key}:{sr_api_secret}",
}

from confluent_kafka.schema_registry import SchemaRegistryClient

client = SchemaRegistryClient(sr_config)
print(client)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/35b454fc-9887-5b47-a593-5130326f44a0.png)

### スキーマの登録と検証

#### スキーマの登録と検証

```python
from confluent_kafka.schema_registry import Schema

avro_str = """
{
  "type": "record",
  "name": "User",
  "namespace": "example.avro",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "favorite_number", "type": "int", "default": 42}
  ]
}
"""

# Schemaオブジェクトを作成 (Avro Schemaの場合は"AVRO"を指定)
schema = Schema(avro_str, "AVRO")
schema
```

```text
Schema(schema_str='\n{\n  "type": "record",\n  "name": "User",\n  "namespace": "example.avro",\n  "fields": [\n    {"name": "name", "type": "string"},\n    {"name": "favorite_number", "type": "int", "default": 42}\n  ]\n}\n', schema_type='AVRO', references=[], metadata=None, rule_set=None)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e2d8d5ed-a8a0-a884-d16c-b090c5e0c964.png)

#### スキーマの登録

```python
subject_name = "sr-test-avro-value"

try:
    schema_id = client.register_schema(subject_name, schema)
    print("Registered schema with ID:", schema_id)
except Exception as e:
    print("Error registering schema:", e)
```

```text
Registered schema with ID: 100001
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6f1f5b8a-9ffa-1f36-5f03-d0aabe6f4b9f.png)

#### スキーマ取得と確認

```python
try:
    fetched_schema = client.get_schema(schema_id)
    print("Fetched schema:", fetched_schema.schema_str)
except Exception as e:
    print("Error fetching schema:", e)
```

```text
Fetched schema: {"type":"record","name":"User","namespace":"example.avro","fields":[{"name":"name","type":"string"},{"name":"favorite_number","type":"int","default":42}]}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f1316877-2c60-4f85-9cf1-1c4c2393b2bc.png)

### スキーマ互換性テスト

スキーマの変更がないため、`True`とリターンされます。

```python
try:
    compatible = client.test_compatibility(subject_name, schema, version='latest')
    print("Is the schema compatible with the latest version?:", compatible)
except Exception as e:
    print("Error testing compatibility:", e)
```

```text
Is the schema compatible with the latest version?: True
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d8f5d0d3-5635-8249-4d77-dc0e45475289.png)

Compatibility が BACKWARD であるため、カラムの追加（`add_col`列の追加）が許容されないため、`False`がリターンされます。

```python
error_avro_str = """
{
  "type": "record",
  "name": "User",
  "namespace": "example.avro",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "favorite_number", "type": "int", "default": 42},
    {"name": "add_col", "type": "string"}
  ]
}
"""

# Schemaオブジェクトを作成 (Avro Schemaの場合は"AVRO"を指定)
error_schema = Schema(error_avro_str, "AVRO")
try:
    compatible = client.test_compatibility(subject_name, error_schema, version='latest')
    print("Is the schema compatible with the latest version?:", compatible)
except Exception as e:
    print("Error testing compatibility:", e)
```

```text
Is the schema compatible with the latest version?: False
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3d7084cb-bea2-941c-092e-536d21769a74.png)


### サブジェクト・バージョン管理操作

#### 全サブジェクト一覧取得

```python
try:
    subjects = client.get_subjects()
    print("All subjects:", subjects)
except Exception as e:
    print("Error getting subjects:", e)
```

```text
All subjects: ['sr-test-avro-value']
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a55331cb-8e0a-77f8-ccbd-5a8bff91351e.png)


#### 指定サブジェクトのバージョン一覧取得

```python
try:
    versions = client.get_versions(subject_name)
    print(f"Versions under {subject_name}:", versions)
except Exception as e:
    print("Error getting versions:", e)
```

```text
Versions under sr-test-avro-value: [1]
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3d33169a-ee02-ee17-0880-5ac1c5fe0a1d.png)

#### 最新バージョン情報取得

```python
try:
    latest_schema_info = client.get_latest_version(subject_name)
    print("Latest schema info:", latest_schema_info)
except Exception as e:
    print("Error getting latest version:", e)
```

```text
Latest schema info: RegisteredSchema(schema_id=100001, schema=Schema(schema_str='{"type":"record","name":"User","namespace":"example.avro","fields":[{"name":"name","type":"string"},{"name":"favorite_number","type":"int","default":42}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None), subject='sr-test-avro-value', version=1)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6f5374c2-4c9e-9022-93a8-8161176f50b8.png)

#### スキーマ互換性レベルの確認

```python
# 現在のグローバル互換性レベル取得
try:
    current_compat = client.get_compatibility()
    print("Current global compatibility level:", current_compat)
except Exception as e:
    print("Error getting compatibility:", e)
```

```text
Current global compatibility level: BACKWARD
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/69bfc865-2916-ab21-5b76-c384f05c0068.png)

#### スキーマ互換性レベルの変更

```python
# グローバル互換性レベルを 'FULL' に変更
try:
    new_level = client.set_compatibility(level="FULL")
    print("New global compatibility level:", new_level)
except Exception as e:
    print("Error setting compatibility:", e)
```

```text
New global compatibility level: {'compatibility': 'FULL'}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1c16d7ee-222e-f97f-6fc8-78b22a8493ef.png)


### エラーハンドリング

#### SchemaRegistryError 

よく発生するエラーには、`SchemaRegistryError` があります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2fd0753c-cb80-2f72-aaf4-e2b4535045f2.png)

下記のコードでは存在しない subject のバージョンを取得しようとして`SchemaRegistryError` を発生させています。

```python
from confluent_kafka.schema_registry import SchemaRegistryError

try:
    # 存在しないサブジェクトを取得してみる
    non_existent_subject = "non-existent-subject"
    non_existent_versions = client.get_versions(non_existent_subject)
except SchemaRegistryError as e:
    print(f"SchemaRegistryError occurred: {e}")
except Exception as e:
    print("Other Error:", e)
```

```text
SchemaRegistryError occurred: Subject 'non-existent-subject' not found. (HTTP status code 404, SR code 40401)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3656aa07-b855-faf4-2cf5-9f88ab3db405.png)


### 事後処理

#### Confluent 上で environment を削除

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4e12b13b-5ca3-2014-ce72-cb71a706de83.png)

## まとめ

Confluent Schema Registry は、Kafka ベースのデータパイプラインにおいてスキーマ管理を標準化・自動化する重要なツールです。これを活用することで、スキーマの変更による障害を回避しやすくなり、堅牢なストリーム処理基盤を構築できます。他社の Schema Registry（Azure や AWS のサービス）とはライセンスや統合度、内部構造などが異なりますが、共通してスキーマ管理の複雑性を軽減する役割を果たします。多様な環境に合わせたスキーマ管理戦略を立てる際、Confluent Schema Registry は有力な選択肢の一つと言えます。