メッセージタイプ
gRPC 経由で Zerobus Ingest SDK を使用して取り込む場合、特定のストリーム上でレコードをどのようにエンコードするかを選択します。Zerobus Ingest は 3 つのメッセージフォーマット (JSON、Protocol Buffers (protobuf)、Apache Arrow) をサポートしているため、ワークロードに合わせてシンプルさ、型安全性、throughput のトレードオフを調整できます。データが永続化される前に、すべてのフォーマットが Delta テーブルのスキーマに対して検証されます。「スキーマの管理」を参照してください。
どの形式を使用すべきですか？
フォーマット | どのようなタスクにベストなのか | 注 |
|---|---|---|
JSON | 導入とシンプルなプロデューサー。 | コンパイルするスキーマ定義を必要としない、最もシンプルなオプションです。便利ですが、大量のワークロードではバイナリ形式よりも低速です。 |
プロトコルのバッファ | 本番運用、行指向、大量のストリーム。 | 型安全でコンパクトなバイナリエンコーディング。ほとんどの本番運用ワークロードに推奨されます。コンパイル済みスキーマが必要です。 |
Apache Arrow | カラムナ型またはバッチ指向のワークロード。 | Apache Arrow レコードバッチを直接送信し、行ごとのシリアル化を回避します。データがすでに列指向である場合、またはバッチで取り込む場合に最適です。ベータ版。Zerobus Ingest での Arrow Flight の使用を参照してください。 |
以下のスニペットは、Python SDKにおける各フォーマットの形状を示しています。これらは、SDKクライアントが既に作成されており、ターゲットテーブルが判明していることを前提としています。完全なセットアップ（Endpoint、テーブル、Service Principal）および各言語の例については、Use Zerobus Ingestを参照してください。
JSON
JSON は最も簡単に起動する方法です。コンパイルするスキーマ定義なしで、レコードを JSON オブジェクトとして送信します。これは、起動、プロトタイピング、および throughput よりも利便性が重視されるプロデューサーに最適です。大量の本番運用ワークロードには、バイナリ形式（protobuf または Arrow）の方が効率的です。
記述子なしで TableProperties にテーブル名を渡すことにより、JSON レコードのストリームを作成します：
table_properties = TableProperties(TABLE_NAME)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)
stream.ingest_record_offset({"device_name": "sensor-1", "temp": 22, "humidity": 55})
JSON の詳細なウォークスルーについては、「クライアントを記述する」を参照してください。
Protocol Buffers
Protobuf は、型安全でコンパクトなバイナリエンコーディングを提供し、ほとんどの本番運用における行指向ワークロードで推奨される形式です。ターゲットの Delta テーブルに適合する protobuf スキーマを定義し（Protobuf スキーマを参照）、それをコンパイルすると、SDK が gRPC を介してレコードごとにデータを取り込みます。
protobuf の使用には 3 つのステップが必要です。テーブルと一致する .proto スキーマを生成し、それを言語モジュールにコンパイルしてから、記述子を TableProperties に渡してレコードを取り込みます。次の例では Python SDK を使用します。
1. テーブルから .proto スキーマを生成します。 Python SDKには、Deltaテーブルを読み取り、一致するスキーマを書き込むgenerate_protoツールが含まれています。
python -m zerobus.tools.generate_proto \
--uc-endpoint "https://<workspace-id>.cloud.databricks.com" \
--client-id "<client-id>" \
--client-secret "<client-secret>" \
--table "main.default.air_quality" \
--output "record.proto" \
--proto-msg "AirQuality"
生成されたスキーマはproto2構文を使用し、各Delta列にオプションのフィールドが含まれます：
syntax = "proto2";
message AirQuality {
optional string device_name = 1;
optional int32 temp = 2;
optional int64 humidity = 3;
}
2. protobuf コンパイラを使用してスキーマを Python モジュールにコンパイルする :
pip install "grpcio-tools>=1.60.0,<2.0"
python -m grpc_tools.protoc --python_out=. --proto_path=. record.proto
これは record_pb2.py を生成します。
3. レコードをインジェスト するには、コンパイルされた記述子を TableProperties (protobuf の default) に渡します。SDK は記述子を使用して各レコードをシリアル化します：
import record_pb2
descriptor_bytes = record_pb2.AirQuality.DESCRIPTOR.file.serialized_pb
table_properties = TableProperties(TABLE_NAME, descriptor_bytes)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)
record = record_pb2.AirQuality(device_name="sensor-1", temp=22, humidity=55)
stream.ingest_record_offset(record)
上記の例では Python SDK を使用しています。ツールは言語によって異なります。一部の SDK にはテーブルから .proto を生成する generate_proto ユーティリティが付属していますが、その他 (Go や TypeScript など) は既存の .proto をコンパイルします。言語ごとのステップについては、「クライアントの作成」の各 SDK tab にある protobuf の注記を参照してください。ツールのソースと完全な例については、Zerobus SDK リポジトリを参照してください。
Apache Arrow
ベータ版
Apache Arrow の取り込みはベータ版です。
Apache Arrowの取り込みでは、各行を最初にJSONやprotobufに変換するのではなく、同じgRPC接続を介して Arrow RecordBatch データを直接送信します。アプリケーションがすでにArrowデータを生成している場合や、行をバッチで取り込む場合に最適な選択肢です。特に、行ごとのシリアル化がオーバーヘッドとなるような、幅が広く数値が多いスキーマや、アナリティクス指向のスキーマに適しています。
Arrowは、非常に大きなバッチにも適しています。すべてか無かであり、メッセージごとのサイズ制限によって制限されるJSONおよびprotobufのバッチメソッドとは異なり、Arrow Flightパスは大きなバッチを小さな転送メッセージに分割し、個別に送信および確認応答を行います。Arrow Flight バッチ are the exceptionおよびUse Arrow Flight with Zerobus Ingestを参照してください。
pyarrow.Schema を使用して Arrow ストリームを開き、RecordBatch データを取り込みます:
stream = sdk.create_arrow_stream(TABLE_NAME, schema, CLIENT_ID, CLIENT_SECRET)
stream.ingest_batch(batch)
スキーマ定義、バッチ処理、圧縮を含む Arrow Flight の詳細なウォークスルーについては、「Zerobus Ingest での Arrow Flight の使用」を参照してください。
関連
- API プロトコル: これらの形式が経由する API プロトコル。
- スキーマ管理：レコードがテーブルに対してどのように検証されるか。
- サポートされているデータ型：サポートされている Delta および Protobuf データ型。