Kafka ClickPipe 向けスキーマレジストリ

ClickPipes は、Avro および Protobuf でエンコードされたレコード値と構造化 Kafka キーをデコードするためのスキーマレジストリとの連携をサポートしています。

Kafka ClickPipes は、2 種類のスキーマレジストリをサポートしています。

• Confluent 互換レジストリ: Confluent スキーマレジストリ自体や Redpanda Schema Registry など、Confluent スキーマレジストリと API 互換性のあるレジストリです。Avro と Protobuf をサポートします。
• AWS Glue Schema Registry: 通常 Amazon MSK から取得される、AWS Glue SerDe でシリアライズされた Avro データ用です。

ClickPipes は現在、Azure Schema Registry をサポートしていません。サポートが必要な場合は、チームまでお問い合わせください。

ClickPipes の設定でスキーマレジストリを統合するには、次のいずれかの方法を使用する必要があります。

1. スキーマの subject への完全なパスを指定します (例: https://registry.example.com/subjects/events)
   • 必要に応じて、URL に /versions/[version] を追加することで特定のバージョンを参照できます (追加しない場合、ClickPipes は最新バージョンを取得します) 。
2. スキーマ ID への完全なパスを指定します (例: https://registry.example.com/schemas/ids/1000)
3. スキーマレジストリのルート URL を指定します (例: https://registry.example.com)

ClickPipes は、指定した URL を使用して HTTPS 経由でスキーマレジストリに接続します。スキーマレジストリをパブリックに公開しておく必要はありません。

Kafka ブローカーに reverse private endpoint (AWS PrivateLink または GCP Private Service Connect) 経由でアクセスしている場合、スキーマレジストリでも同じプライベート接続を利用できます。ClickPipes は reverse private endpoint のプライベート DNS を使ってレジストリのホスト名を名前解決するため、ブローカーと同じくプライベート環境でホストされているレジストリであっても、そのホスト名が reverse private endpoint のプライベート IP アドレスに解決される限り到達できます (エンドポイントのプライベート DNS サポート、または カスタムのプライベート DNS マッピング を使用) 。

次の点に注意してください。

• スキーマレジストリの URL には https:// を使用する必要があります。
• レジストリのホスト名がプライベートアドレスに解決される場合は、ClickPipe 用に選択された reverse private endpoint 経由で到達可能である必要があります。そうでないと、Setup 中の接続チェックは失敗します。

ClickPipes は、設定されたスキーマレジストリからスキーマを動的に取得して適用します。

• レコード値にスキーマ 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 SerDe を使用して Avro をシリアル化する場合 (たとえば、Amazon MSK トピックに対して AWSKafkaAvroSerializer を使用する場合) 、ClickPipes は AWS Glue Schema Registry からこれらのスキーマを直接解決できます。Glue は Confluent 互換レジストリとは異なるワイヤ形式と API を使用するため、個別に設定する必要があります。

AWS Glue Schema Registry の設定は現在、ClickHouse Cloud console でのみ利用できます。ClickPipes API または Terraform プロバイダーではサポートされていません。

ClickPipe 作成ウィザードの Kafka 接続ステップで スキーマレジストリ を有効にし、レジストリタイプ を AWS Glue に設定します。

設定するレジストリ URL はありません。Glue SerDe で生成された各レコードには、そのスキーマバージョンの ID が含まれます。ClickPipes は glue:GetSchemaVersion を使用してスキーマバージョンを解決し、異なるスキーマバージョンごとに API 呼び出しを 1 回行ってキャッシュします。スキーマ進化は自動的に処理されます。ストリームの途中でレコードが新しいスキーマバージョンに切り替わった場合は、最初に検出された時点で解決されます。

環境に合った方のオプションを使用してください。Amazon MSK では通常、オプション A を使用します。

Kafka ClickPipe がすでに IAM を使用して MSK に認証している場合、ClickPipes は同じ IAM ID を使用してレジストリを読み取ります。IAM ロール ARN フィールドは空のままにして、次のステートメントを ID の権限に追加します。

• IAM ロール: MSK 用に設定されたロールの権限ポリシーにステートメントを追加します。
• IAM 認証情報: アクセスキーに関連付けられた IAM プリンシパルの権限ポリシーにステートメントを追加します。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ClickPipesGlueSchemaRegistryRead",
      "Effect": "Allow",
      "Action": ["glue:GetSchemaVersion"],
      "Resource": "*"
    }
  ]
}

ロールベースの認証では、信頼ポリシーを変更する必要はありません。MSK 用に設定されている信頼関係が、このアクセスをすでにカバーしています。IAM 認証情報では、ロールの信頼ポリシーは使用しません。

ブローカーが IAM で認証されない場合 (SASL/SCRAM、SASL/PLAIN、mTLS) 、またはレジストリとブローカーが異なる AWS アカウントにある場合に使用します。

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 サービスには固有のロールがあるため、この値はサービスごとに異なります。

レジストリ IAM ロールを作成する

AWS アカウントに IAM ロールを作成します。ロール名は必ず ClickHouseAccessRole- で始める必要があります。

信頼ポリシーを設定する

{ClickHouse_IAM_ARN} を前のステップで取得した値に置き換えます。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "{ClickHouse_IAM_ARN}"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

アクセス許可ポリシーを設定する

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ClickPipesGlueSchemaRegistryRead",
      "Effect": "Allow",
      "Action": ["glue:GetSchemaVersion"],
      "Resource": "*"
    }
  ]
}

ClickPipe を設定する

新しいロールの ARN をウィザードの IAM ロール ARN フィールドに貼り付けます。

• Avro のみをサポートします。Glue SerDe 経由の JSON Schema および Protobuf はサポートされていません。
• Kafka ログソースのみをサポートします。Kinesis ClickPipes では Glue registry を使用できません。

次のルールは、Confluent 互換レジストリと AWS Glue Schema Registry の両方に適用されます。これらのルールは、取得した値のスキーマと ClickHouse の宛先テーブルのマッピングを規定し、_key. プレフィックスを持つ構造化キーからマッピングされたレコードまたはメッセージのフィールドにも適用されます。

• スキーマに、ClickHouse の宛先マッピングに含まれていないフィールドがある場合、そのフィールドは無視されます。
• スキーマに、ClickHouse の宛先マッピングで定義されているフィールドがない場合、ClickHouse のカラムには 0 や空文字列などの「ゼロ」値が設定されます。なお、DEFAULT 式はサポートされていません。
• スキーマのフィールドと ClickHouse のカラムに互換性がない場合、その行/メッセージの挿入は失敗し、その失敗は ClickPipes のエラーテーブルに記録されます。なお、いくつかの暗黙的な変換 (たとえば数値型同士) はサポートされていますが、すべてがサポートされるわけではありません (たとえば、Avro のレコードフィールドを Int32 の ClickHouse カラムに挿入することはできません) 。