Kafka ClickPipes でサポートされるスキーマレジストリ
- Confluent 互換レジストリ: Confluent スキーマレジストリ自体や Redpanda Schema Registry など、Confluent スキーマレジストリと API 互換性のあるレジストリです。Avro と Protobuf をサポートします。
- AWS Glue Schema Registry: 通常 Amazon MSK から取得される、AWS Glue SerDe でシリアライズされた Avro データ用です。
Confluent 互換レジストリ
設定
- スキーマの subject への完全なパスを指定します (例:
https://registry.example.com/subjects/events)- 必要に応じて、URL に
/versions/[version]を追加することで特定のバージョンを参照できます (追加しない場合、ClickPipes は最新バージョンを取得します) 。
- 必要に応じて、URL に
- スキーマ ID への完全なパスを指定します (例:
https://registry.example.com/schemas/ids/1000) - スキーマレジストリのルート URL を指定します (例:
https://registry.example.com)
ネットワーク接続
- スキーマレジストリの URL には
https://を使用する必要があります。 - レジストリのホスト名がプライベートアドレスに解決される場合は、ClickPipe 用に選択された reverse private endpoint 経由で到達可能である必要があります。そうでないと、Setup 中の接続チェックは失敗します。
仕組み
- レコード値にスキーマ ID が埋め込まれている場合は、それを使ってスキーマを取得します。
- レコード値にスキーマ ID が埋め込まれていない場合は、ClickPipe の設定で指定されたスキーマ ID または subject 名を使ってスキーマを取得します。
- レコード値に埋め込みのスキーマ ID がなく、かつ ClickPipe の設定でスキーマ ID も subject 名も指定されていない場合、スキーマは取得されず、ClickPipes のエラーテーブルに
SOURCE_SCHEMA_ERRORが記録され、そのメッセージはスキップされます。 - レコード値がスキーマに準拠していない場合、ClickPipes のエラーテーブルに
DATA_PARSING_ERRORが記録され、そのメッセージはスキップされます。 - Protobuf スキーマのみ: ClickPipes は、依存関係として定義された import 済みのスキーマを読み込みます。外部参照を含む Avro スキーマは、現時点ではサポートされていません。
_key.id などのフィールドのマッピングが設定されている場合、ClickPipes は Kafka のキーに埋め込まれたスキーマ ID をレコード値とは独立して解決します。キーには異なるスキーマ ID を使用できますが、値と同じレジストリファミリーおよびシリアライゼーションフォーマットを使用する必要があります。解決されたキースキーマはキャッシュされ、スキーマ変更は自動的に検出されます。
AWS Glue Schema Registry
AWSKafkaAvroSerializer を使用する場合) 、ClickPipes は AWS Glue Schema Registry からこれらのスキーマを直接解決できます。Glue は Confluent 互換レジストリとは異なるワイヤ形式と API を使用するため、個別に設定する必要があります。
AWS Glue Schema Registry の設定は現在、ClickHouse Cloud console でのみ利用できます。ClickPipes API または Terraform プロバイダーではサポートされていません。
Avro のみ。 AWS Glue レジストリは Avro フォーマットでのみサポートされています。Glue SerDe は JSON と Protobuf のフレーミングにも対応していますが、ClickPipes ではサポートされておらず、パイプの作成時に拒否されます。
設定
設定するレジストリ URL はありません。Glue SerDe で生成された各レコードには、そのスキーマバージョンの ID が含まれます。ClickPipes は
glue:GetSchemaVersion を使用してスキーマバージョンを解決し、異なるスキーマバージョンごとに API 呼び出しを 1 回行ってキャッシュします。スキーマ進化は自動的に処理されます。ストリームの途中でレコードが新しいスキーマバージョンに切り替わった場合は、最初に検出された時点で解決されます。
IAM の設定
オプション A: ブローカーの IAM ID を再利用する
- IAM ロール: MSK 用に設定されたロールの権限ポリシーにステートメントを追加します。
- IAM 認証情報: アクセスキーに関連付けられた IAM プリンシパルの権限ポリシーにステートメントを追加します。
オプション B: 専用のレジストリロールを使用する
AWS デプロイメントのみ。 このオプションはサービスの AWS IAM ロールを利用するため、AWS にデプロイされた ClickHouse Cloud サービスが必要です。サービスが GCP または Azure 上で稼働しており、ブローカーが IAM 認証を使用しない場合は、専用のレジストリロールを設定できません。
1
ClickHouse サービスの IAM ロール ARN を取得する
サービスを開き、Settings タブを選択します。Network security information セクションまでスクロールし、
arn:aws:iam::123456789012:role/CH-S3-example-service-Role 形式の ARN である サービスロール ID (IAM) の値をコピーします。以降、この値を {ClickHouse_IAM_ARN} と呼びます。AWS にデプロイされた各 ClickHouse サービスには固有のロールがあるため、この値はサービスごとに異なります。2
レジストリ IAM ロールを作成する
AWS アカウントに IAM ロールを作成します。ロール名は必ず アクセス許可ポリシーを設定する
ClickHouseAccessRole- で始める必要があります。信頼ポリシーを設定する{ClickHouse_IAM_ARN} を前のステップで取得した値に置き換えます。3
ClickPipe を設定する
新しいロールの ARN をウィザードの IAM ロール ARN フィールドに貼り付けます。
IAM リソースのスコープ。 これらの例は、どちらも
"*" に対する glue:GetSchemaVersion を許可する、AWS のデシリアライザー向けドキュメント化されたポリシーと、その AWSGlueSchemaRegistryReadonlyAccess マネージドポリシーに従っています。ClickPipes は、解決された各スキーマが設定した レジストリ名 に一致するかを個別に確認し、他のレジストリのバージョンは拒否します。トラブルシューティング
制限事項
- Avro のみをサポートします。Glue SerDe 経由の JSON Schema および Protobuf はサポートされていません。
- Kafka ログソースのみをサポートします。Kinesis ClickPipes では Glue registry を使用できません。
スキーママッピング
_key. プレフィックスを持つ構造化キーからマッピングされたレコードまたはメッセージのフィールドにも適用されます。
- スキーマに、ClickHouse の宛先マッピングに含まれていないフィールドがある場合、そのフィールドは無視されます。
- スキーマに、ClickHouse の宛先マッピングで定義されているフィールドがない場合、ClickHouse のカラムには 0 や空文字列などの「ゼロ」値が設定されます。なお、
DEFAULT式はサポートされていません。 - スキーマのフィールドと ClickHouse のカラムに互換性がない場合、その行/メッセージの挿入は失敗し、その失敗は ClickPipes のエラーテーブルに記録されます。なお、いくつかの暗黙的な変換 (たとえば数値型同士) はサポートされていますが、すべてがサポートされるわけではありません (たとえば、Avro のレコードフィールドを
Int32の ClickHouse カラムに挿入することはできません) 。