メインコンテンツまでスキップ
に最終更新

スキーマ管理

この詳細解説では、Lakeflow Connect の Zerobus Ingest がどのように受信レコードを Delta テーブルスキーマと照合して検証するか、また部分データや進化するデータに対してどのようにスキーマを設計するかを説明します。

多くのプロデューサーが同じ Delta テーブルに書き込みを行い、Zerobus Ingest は各レコードを1つの固定されたテーブルスキーマに対して検証します。レコードは全体として承認または却下され、準拠していないフィールドは、設定されている場合にレスキュー VARIANT 列にキャプチャされます。

テーブルが契約となります​

Your Delta table schema is the authoritative contract for what Zerobus Ingest accepts.Zerobus Ingest はその契約に基づいてすべてのレコードをゲートしますが、契約をどれだけ厳格にするか、あるいはどれだけ許容するかはユーザーが選択できます。同じサービスであっても、テーブルの定義方法に応じて、厳格なスキーマを適用したり、柔軟な列のサブセットを受け入れたり、適合しないすべてをキャッチしたりすることができます。

  • Zerobus Ingestがデータをゲートします。 すべてのレコードをターゲットテーブルと照らし合わせて検証し、適合しないものはすべて拒否します。推測や列のサイレント削除を行うことはありません。
  • コントラクトを定義します。 列を必須または null 許容としてマークし、レスキュー列を追加することで、何が「適合」するかを決定します。
  • Zerobus Ingest がテーブルを拡張することはありません。 レコードに対応するために列を追加したり、型を変更したり、スキーマを進化させたりすることはありません。Zerobus Ingest が受け入れる内容は、テーブルを進化させることで進化させます。その逆ではありません。

次のセクションでは、そのコントラクトを形成する 3 つの方法を、最も許容度の高いものから厳格なもの、包括的なものまで順に示します。

レコードがテーブルに照合される仕組み​

レコードは宛先テーブルに適合する必要があります。つまり、テーブル内のすべての null 非許容列を最低限含んでいる必要があります。テーブルで null 値を許容する列はレコードから省略でき、NULL として書き込まれます。null 値を許容する列の省略は破壊的でない変更として扱われるため、テーブルに null 値を許容する列を追加しても、それらを含まない古いレコードの取り込みを継続できます。

レコードがテーブルに適合しない場合、Zerobus Ingest はエラーを返します。これには以下が含まれます。

  • NULL 非許容の列が不足しています。
  • Delta テーブルに存在しない列名（レスキュー列を設定している場合を除く）。
  • Delta テーブルと互換性のない型の列。サポートされている Delta および Protobuf データ型については、サポートされているデータ型を参照してください。

コントラクトを形成する3つの方法​

テーブルをどのように定義するかによって、取り込みの厳格さや許容度が決まります。以下の3つのシナリオは、最も許容度が高いものから、厳格なもの、すべてを網羅するものへと順に並んでいます。

シナリオ 1: すべての列がオプション（サブセットを受け入れる）​

すべての列をNULL可にします。プロデューサーは列の任意のサブセットを送信でき、省略された列は NULL として書き込まれます。テーブルに存在しない列を含むレコードは、引き続き拒否されます。

SQL
CREATE TABLE main.default.air_quality (
  device_name STRING,
  temp INT,
  humidity INT);
  • {"device_name": "sensor-1", "temp": 22, "humidity": 55}: 承認されました。 すべての列が存在します。
  • {"device_name": "sensor-1"}: 承認済み。 temp と humidity は NULL 可であるため、NULL と記述されます。
  • {"device_name": "sensor-1", "temp": 22, "region": "us-west"}: 却下されました。 region はテーブル内に存在しません。

シナリオ 2: 必須列 (特定のフィールドを強制)​

列 NOT NULL をマークして必須にします。すべてのレコードでそれらの列を提供する必要があります。そうでない場合は拒否されます。これはスペクトルの厳格な端にあたります。フィールドが常に存在する必要がある場合に使用してください。

SQL
CREATE TABLE main.default.air_quality (
  device_name STRING NOT NULL,
  temp INT NOT NULL,
  humidity INT);
  • {"device_name": "sensor-1", "temp": 22, "humidity": 55}: 承認されました。 必要な列はすべて存在します。
  • {"device_name": "sensor-1", "temp": 22}: 承認済み。 humidity は NULL 可であるため、NULL として書き込まれます。
  • {"device_name": "sensor-1"}: 却下されました。 temp は NULL 非許容ですが、欠落しています。

シナリオ 3: レスキュー列（その他すべてをキャッチ）​

レコードを拒否する代わりに、スキーマに適合しないフィールドをキャプチャするための VARIANT レスキュー列を追加します。テーブルに一致するフィールドは、通常どおりそれぞれの列に書き込まれます。余分なフィールドや準拠していないフィールドは、JSON オブジェクトとしてレスキュー列にグループ化されます。これは最も許容範囲が広い設定です。余分なフィールドや null 許容列の型の不一致は、拒否されるのではなくキャプチャされます。必須（null 非許容）列が省略されている場合、レスキュー列はスキーマが必要とする値を提供できないため、レコードは引き続き拒否されます。レスキュー列は Beta 版であり、現在は JSON 形式の取り込みをサポートしています。

  • {"device_name": "sensor-1", "temp": 22, "region": "us-west"}: 承認されました。 region はテーブルに存在しないため、拒否される代わりにレスキュー列にキャプチャされます。

レスキュー列の設定方法と正確なルールについては、Zerobus レスキュー列を参照してください。

スキーマ進化​

Zerobus Ingest は、ターゲットテーブルを自動的に進化させません。データの形状が変更された場合は、最初にテーブルを進化させ（例：ALTER TABLEを使用）、その後新しいスキーマに対してレコードを送信してください。

null 値を許容する列の追加は破壊的でない変更です。新しい列を送信しない既存のプロデューサーはそのまま動作し続け、そのレコードには NULL が設定されます。これにより、スキーマの変更とプロデューサーの変更を個別にロールアウトできます。

Protobufスキーマ​

Protocol Buffers (protobuf) を使用して取り込む場合、protobuf メッセージ定義にも同じ適合ルールが適用されます。つまり、Delta テーブル内のすべての null 非許容列を最低限含める必要があり、null 許容列は省略可能です。

以下は、protobuf スキーマにも適用されます。

  • Zerobus Ingest は、2000 列を超える proto スキーマをサポートしていません。
  • Zerobus Ingest は、ASCII 文字、数字、アンダースコアを使用したテーブル名と列名のみをサポートしています。
  • Zerobus Ingest は、「ストリーム作成」および「レコード取り込み」操作で異なる proto スキーマを使用することをサポートしていません。
このページの見出し