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

Zerobus Ingest のコンセプト

このページでは、Lakeflow Connect における Zerobus Ingest の主要な概念（サービスの仕組み、ストリーム、サーバー、クライアント、およびサポートされているデータ型）について説明します。

コンセプトにジャンプ:

Zerobus Ingest の仕組み​

データプロデューサーは、まず Zerobus Ingest API に接続してターゲットの Delta テーブルを指定し、そのスキーマに一致するメッセージを構築してから、メッセージを送信します。サービスはデータを耐久性のあるものにし、耐久性の確認を提供するインターフェイスについては、クライアントのメッセージを確認します。その後、別のステップとして、最適化された方法でデータを Delta テーブルにマテリアライズします。耐久性の確認は、クエリー照会可能性ではなく、耐久性を確認するものです。これがどのように機能するか、およびクライアントにとってそれが何を意味するかについては、非同期通信を参照してください。

Zerobus Ingest は、ワークロードに合わせて弾力的にスケーリングする Serverless サービスです。スケーリングの詳細については、以下の How Zerobus Ingest scales を参照してください。

Zerobus Ingestの仕組み

このセクションでは、Zerobus Ingest への接続方法と、データの形式についても説明します。

  • API プロトコル: API プロトコル、SDK 付き gRPC、REST、OpenTelemetry、MQTT、および Kafka 互換 API、およびそれぞれの使用時期。
  • メッセージ型: レコード形式である JSON、プロトコル バッファー (protobuf)、および Apache Arrow と、それぞれの使用場面について。

サーバー​

Zerobus Ingest サービスは、テーブルを自動的に作成または操作しません。ユーザー自身がテーブルを作成する必要があります。テーブルとそのスキーマは、受信データの期待値に関する信頼できるソースです。

Zerobus Ingest サーバーは、クライアントから送信されたデータを受け入れ、それがターゲットテーブルのスキーマに適合していることを検証します。レコードが適合する場合、サーバーはそのレコードを耐久性のあるものにし、インターフェイスが耐久性の確認応答を提供する場合は、クライアントに応答を返します。レコードを Delta テーブルにマテリアライズしてクエリ可能にすることは、その直後の別のステップとして行われます。

サービス責任には以下が含まれます：

  • テーブルに対するメッセージのスキーマ検証。
  • レコードを耐久性のあるものにし、インターフェイスが耐久性の確認を提供するときに、それをクライアントに通知します。この確認は、耐久性を示すものであり、レコードがまだクエリー照会可能であることを示すものではありません。
  • ターゲットテーブルへのデータのマテリアライズをタイムリーに行うことで、クエリが可能になります。レイテンシーの数値については、レイテンシーを参照してください。

クライアント​

クライアントは Zerobus Ingest に接続してレコードを送信し、インターフェイスがサポートしている場合は、それらが永続的であることを確認します。Zerobus Ingest SDK を使用する場合、SDK がこれらの処理のほとんどを自動的に行うため、設定する内容と SDK が自動的に行う処理を分けて理解しておくと便利です。

構成または実装するもの：

  • ターゲットテーブルの選択。
  • Zerobus Ingestサービスへのストリームを開いています。
  • スキーマ互換性のあるメッセージを構築して送信します。

SDK は自動的に以下を処理します：

  • メッセージの承諾。 SDKが代わりに承諾ループを実行し、オフセットまたは 承諾コールバック を通じて耐久性の確認を提示します。アプリケーションが必要とする場合にのみ、特定のレコードでブロックします。非同期通信を参照してください。
  • リカバリ。 defaultでは、一時的なエラーが発生した場合、SDK は再接続を行い、未確認のレコードをリプレイします。
    • 組み込みのリカバリをオフにして、代わりに独自のリカバリメカニズムを実装することもできます。リカバリを Trigger するもの、構成オプション、およびカスタムリカバリパターンについては、「Recovery and retry patterns」を参照してください。

SDK を使用する場合、確認ロジックやリカバリロジックを手動で記述する必要はありません。SDK を使用しないカスタム統合の場合、 Zerobus SDK リポジトリが統合構造とリカバリ処理の参照先となります。

Stream​

ストリームは、クライアントと Zerobus Ingest サーバー間の直接接続であり、永続的な双方向 gRPC 接続を介して確立されます。SDK はストリームを使用して、長期間持続する高 throughput の接続を促進します。

  • ストリームは、SDK を使用した gRPC API でのみ使用されます。
  • ストリームは、単一のターゲットテーブルにデータを取り込みます。
  • 追加のストリームを開いて異なるテーブルに書き込むか、ワークロードの要件に合わせて単一クライアントのthroughputを拡張します。

ストリームは順序付けの単位でもあり（順序の保証を参照）、Zerobus Ingest がスケールする単位でもあります（Zerobus Ingest のスケーリング方法を参照）。

順序の保証​

順序はストリームごとに保証されます。レコードは、単一のストリームでエンキューされた順序でターゲットテーブルにコミットされます。ストリーム間でのグローバルな順序付けはありません。これに基づき、いくつかの設計上のポイントがあります。

  • レコードを複数のストリームに分散させた場合（ラウンドロビンなど）、それらのストリーム間での順序は保証されません。
  • ユースケースで多数のプロデューサーやストリーム全体にわたる単一の全順序が必要な場合は、取り込み順序に依存するのではなく、アプリケーションでその順序を強制してください（例えば、クエリー対象のTimestampやシーケンス番号を使用します）。

gRPC ストリーミングの理由​

ストリームの gRPC 接続が維持されるため、クライアントはステアレスプロトコルのリクエストごとのセットアップコストを回避し、単一のチャンネルを通じて継続的かつ大容量のレコードフローをプッシュできます。これにより、SDK が最も throughput の高い取り込み方法となっています。その他のインターフェイス（REST、OpenTelemetry、MQTT、および Kafka 互換 APIs）とそれぞれの選択基準については、API プロトコルを参照してください。

Zerobus Ingest のスケーリング方法​

Zerobus Ingest は高いスケーラビリティを実現するように設計されており、容量計画を立てることなくその規模に到達できます。これを可能にする 2 つの設計上の選択肢があります。

  • Serverlessです。 このサービスは負荷の変化に応じて自動的に容量を追加および削除するため、ブローカーのサイズ設定やパーティションのプロビジョニングを行う必要はありません。ワークロードが必要とする数だけ、並列ストリームを開き、テーブルに書き込むことができます。
  • ストリームは動的なパーティション単位です。 スケールアウトのために再パーティション化や再調整が必要となる固定されたパーティションセットとは異なり、ストリームは開いたり、閉じたり、ローテーションしたりすることができます。ストリームをローテーションすることで、需要の変化に応じてサービスが容量とリソースを再調整できるようになります。そのため、ストリームを増やしてプロデューサーを多く実行することでスケールし、残りの処理はサービス側で吸収されます。

実質的な結果として、「hello world」クライアントとペタバイト規模のワークロードは、本質的に同じコードを実行します。違いは、実行するプロデューサーとストリームの数です。この設計により、単一の Delta テーブルへの1兆件を超えるレコードの取り込みが維持されています。技術的な背景については、Ingesting the Milky Way: Petabyte-Scale with Zerobus Ingest ブログ記事を参照してください。

テーブル要件​

Zerobus Ingest は、ユーザーが作成および所有する Delta テーブルに書き込みを行います。ターゲットテーブルとワークスペースは、以下の要件を満たしている必要があります。

  • ターゲットテーブルは、管理対象の Delta テーブルまたはストリーミングテーブルである必要があります。
  • テーブル名には、ASCIIの文字、数字、およびアンダースコアのみを使用する必要があります。
  • ワークスペースとターゲットテーブルは、両方ともサポートされているリージョンのいずれかにある必要があります。

テーブルスキーマに対してレコードがどのように検証されるかについては、スキーマ管理を参照してください。パーティション化やリキッドクラスタリングなどのテーブル機能については、Deltaテーブル機能を参照してください。サポートされていないテーブルタイプと操作については、制限事項を参照してください。

制限事項​

Zerobus Ingest には以下の制限があります。

  • Zerobus Ingestは、外部テーブルやビューには書き込みません。

  • Zerobus Ingest は、プライベート Endpoint を介して保護されたストレージには書き込みません。

  • Zerobus Ingestは、ターゲットテーブルを作成したり、そのスキーマを変更したりしません。テーブルを作成し、そのスキーマを進化させます。スキーマ進化を参照してください。

  • Zerobus Ingestはターゲットテーブルの再作成をサポートしていません。

  • Zerobus Ingest は、取り込みの有効期間中のターゲットテーブルの名前変更または置換をサポートしていません。ターゲットテーブルの名前変更または置換を参照してください。

  • Zerobus Ingestでは、ASCIIの文字、数字、アンダースコア以外の文字を含むテーブル名はサポートされていません。

  • Zerobus Ingest はストリーム間の順序付けを保証しません。順序付けは単一のストリーム内でのみ保証されます。「順序保証」を参照してください。

  • Zerobus Ingestは、defaultストレージにバックアップされたテーブルに書き込みません。

throughput、レコードサイズ、パーティションファンアウトなどの数値の制限については、Zerobus Ingest のクォータを参照してください。

ターゲットテーブルの名前の変更または置き換え​

ストリームによる書き込みが行われている間は、ターゲットテーブルの名前の変更、削除と再作成、またはビューなどの別のオブジェクトへの置き換えを行わないでください。Zerobus Ingestは、名前が変更されたり置き換えられたりしたテーブルにバッファリングされたデータをフラッシュできないため、影響を受けるストリームの進行が停止し、まだ確認されていないレコードをフラッシュできなくなります。フラッシュされないままのデータは、バッファリングされたデータの保持期間の経過後に削除されます。

ALTER TABLE による列の追加などのスキーマ変更がサポートされています。スキーマ進化を参照してください。

ターゲットテーブルの名前を変更または移行するには：

  1. プロデューサーを停止し、テーブルに書き込むすべてのストリームをフラッシュしてクローズします。フラッシュと正常なクローズを参照してください。
  2. 最新のレコードがテーブルに表示されていることを確認します。
  3. テーブルの名前を変更または移行します。
  4. データの取り込みを行うService Principalに、新しいテーブルへの USE CATALOG、USE SCHEMA、SELECT、および MODIFY を付与します。Service Principalの作成と権限の付与を参照してください。
  5. プロデューサーを更新して新しいテーブル名を使用するように設定してから、新しいストリームを開きます。プロデューサーがテーブルにスコープされたOAuthトークンをリクエストする場合は、新しいテーブル用にもリクエストします。

サポートされているデータ型​

次の表は、取り込み用にサポートされている Delta タイプと、それに対応する Protobuf タイプを示しています。

Delta 型

Protobuf 型

INTEGER

int32

STRING

string

FLOAT

float

LONG

int64

SHORT

int32

DOUBLE

double

DECIMAL(p, s)

Decimalテキスト（例：）「123.45」、「1e2」など。

string

BOOLEAN

bool

BINARY

bytes

BYTE (TINYINT)

int32

DATE

int32 （エポックからの日数）に変換する必要があります。

int32

TIMESTAMP

int64 (エポック時間、マイクロ秒単位) に変換する必要があります。

int64

TIMESTAMPNTZ

int64 (エポック時間、マイクロ秒単位) に変換する必要があります。

int64

ARRAY<TYPE>

repeated TYPE

MAP<K,V>

map<K,V>

map Protobuf の糖衣構文（syntactic sugar）は、Protobuf コンパイラのバージョン 3 以降でのみ利用可能です。

STRUCT<FIELDS>

message Nested { FIELDS }

VARIANT

gRPC SDK および REST を介して、STRING 型のキーを持つ JSON エンコードされた文字列としてバリアント値を取り込みます。Zerobus Ingest は、そのデータをシュレッディングせずに列に書き込みます。Apache Arrow Flight の場合、クライアントは代わりにバリアント列の基盤となる metadata フィールドと value フィールドを構築します。VARIANT 列の取り込みを参照してください。

サポートされている形式は以下の通りです：

  • オブジェクト： "{\"id\":0,\"example\":\"this is variant example\"}"
  • プリミティブ: "5"、"3.14"、 "\"string\""
  • 配列: "[1,2,3]"

string

Delta 型

Protobuf 型

INTEGER

int32

STRING

string

FLOAT

float

LONG

int64

SHORT

int32

DOUBLE

double

DECIMAL(p, s)

Decimalテキスト（例：）「123.45」、「1e2」など。

string

BOOLEAN

bool

BINARY

bytes

BYTE (TINYINT)

int32

DATE

int32 （エポックからの日数）に変換する必要があります。

int32

TIMESTAMP

int64 (エポック時間、マイクロ秒単位) に変換する必要があります。

int64

TIMESTAMPNTZ

int64 (エポック時間、マイクロ秒単位) に変換する必要があります。

int64

ARRAY<TYPE>

repeated TYPE

MAP<K,V>

map<K,V>

map Protobuf の糖衣構文（syntactic sugar）は、Protobuf コンパイラのバージョン 3 以降でのみ利用可能です。

STRUCT<FIELDS>

message Nested { FIELDS }

VARIANT

gRPC SDK および REST を介して、STRING 型のキーを持つ JSON エンコードされた文字列としてバリアント値を取り込みます。Zerobus Ingest は、そのデータをシュレッディングせずに列に書き込みます。Apache Arrow Flight の場合、クライアントは代わりにバリアント列の基盤となる metadata フィールドと value フィールドを構築します。VARIANT 列の取り込みを参照してください。

サポートされている形式は以下の通りです：

  • オブジェクト： "{\"id\":0,\"example\":\"this is variant example\"}"
  • プリミティブ: "5"、"3.14"、 "\"string\""
  • 配列: "[1,2,3]"

string

Avro 型​

備考

ベータ版

この機能はベータ版です。使用するには、ワークスペース管理者は プレビュー ページから Zerobus Ingest Avro Support を有効にする必要があります。Databricksのプレビューを管理を参照してください。

Avro フォーマットのレコードを取り込むと、Zerobus Ingest は、ユーザーが提供する Avro ライター スキーマに対して各レコードをデコードし、ターゲット Delta テーブルの型に変換します。次の表は、Avro 型が Delta 型にどのようにマップされるかを示しています。

Avro型

Delta型

boolean

BOOLEAN

int

INT

long

BIGINT

float

FLOAT

double

DOUBLE

string

STRING

bytes

BINARY

record

STRUCT

enum

STRING

array

ARRAY

map

キーはstringでなければなりません。

MAP

fixed uuid 論理型 (16バイト)

STRING

fixed （その他）

BINARY

date （論理型）

DATE

timestamp-millis または timestamp-micros (論理型)

TIMESTAMP

decimal （論理型）

精度とスケールは Avro decimal 型から取得されます。

DECIMAL(p, s)

union ["null", T]

T (NULL可)

union ["int", "long"]

BIGINT

union ["float", "double"]

DOUBLE

union (他のマルチタイプ)

STRUCT null以外の各Branchに対して、member0、member1などのフィールドを含む

Avro型

Delta型

boolean

BOOLEAN

int

INT

long

BIGINT

float

FLOAT

double

DOUBLE

string

STRING

bytes

BINARY

record

STRUCT

enum

STRING

array

ARRAY

map

キーはstringでなければなりません。

MAP

fixed uuid 論理型 (16バイト)

STRING

fixed （その他）

BINARY

date （論理型）

DATE

timestamp-millis または timestamp-micros (論理型)

TIMESTAMP

decimal （論理型）

精度とスケールは Avro decimal 型から取得されます。

DECIMAL(p, s)

union ["null", T]

T (NULL可)

union ["int", "long"]

BIGINT

union ["float", "double"]

DOUBLE

union (他のマルチタイプ)

STRUCT null以外の各Branchに対して、member0、member1などのフィールドを含む

このマッピングは、Spark Avro データソースの規約に従います。Avro データ型システム自体については、Apache Avro 仕様を参照してください。

TIMESTAMP 列はUTCインスタントとして書き込まれます。Avroには8ビットまたは16ビットのInteger型がないため、TINYINT 列と SMALLINT 列はAvroからターゲットにできません。AvroもVARIANT、VOID、INTERVAL、GEOMETRY、またはGEOGRAPHY列には書き込めません。これらのいずれかをターゲットとするスキーマは、ストリームの作成時に拒否されます。

Avroライターのスキーマでも、各型をインラインで宣言する必要があります。名前付き型の参照はサポートされていません。Avroを参照してください。

このページの見出し