メッセージのブロックと確認応答
The Zerobus Ingest SDKs にはレコードをインジェストするためのいくつかのメソッドが用意されており、throughput と受け取る耐久性の確認量との間でトレードオフが生じます。このページでは、各メソッドと、いつ耐久性に関してブロックすべきかを説明します。ブロックせずに非同期で確認応答に対応するには、確認コールバックを参照してください。
このページの例では、Python SDK を使用しています。各メソッドが受け入れる正確なタイムアウトおよび構成オプション（default値や単位を含む）については、Zerobus SDK リポジトリを参照してください。他の言語の SDK でも、同等のオプションが公開されています。
オフセットとは？
取り込まれるすべてのレコードにはオフセットが割り当てられます。これは ストリーム 内での位置を示します。オフセットとは、レコードが永続的に書き込まれたことを確認したい場合に、特定のレコードを参照する方法です。Zerobus Ingest は「少なくとも1回」の配信保証を提供しており、オフセットを待機することは、クライアントが特定のレコードに対してその保証を確認する方法となります。
オフセットの確認は、レコードが永続化されたことを意味しますが、Delta テーブルでクエリ可能になったことを意味するわけではありません。Zerobus Ingest は、その後すぐに永続的なレコードをテーブルにマテリアライズします。レイテンシーの数値については、レイテンシーを参照してください。
データ取り込み方法
SDK には、レコードをインジェストする方法が 2 つ用意されています。(以下のメソッド名は Python SDK のものです。他の SDK でも同等のメソッドが公開されています。)
手法 | 戻り値 | 使用する場合: |
|---|---|---|
オフセットベース 、 | ストリーム上でレコードがキューに入れられた後の、そのレコードのオフセット。 | Recommended default. レコードを順番にエンキューし、必要に応じて後でオフセットを待機することで耐久性を確認したい場合。 |
フューチャーベース 、 | 待機可能な | 非推奨。パフォーマンスを向上させるには、オフセットベースの使用を推奨します。 |
オフセットベース（推奨）
ingest_record_offset() レコードを送信し、そのレコードがストリーム上でキューに入れられるとオフセットを返します。この呼び出しは呼び出し元のスレッドで実行されるため、レコードはメソッドを呼び出した順序でキューに入れられます。また、返されたオフセットを使用して、後で wait_for_offset() で耐久性を確認できます。これはほとんどのプロデューサーで推奨される default であり、 Use Zerobus Ingest の例で使用されている方法です。
将来ベース（非推奨）
ingest_record() 耐久性が確保されるまで待機できる RecordAcknowledgment オブジェクトを返します。これは廃止され、よりパフォーマンスの高いオフセットベースのメソッドが採用されました。まだ移行されていない既存のコードに対してのみ使用してください。
レコードごとの取り込みとバッチ取り込みの比較
各取り込み方法には、1回の呼び出しでレコードのリストを送信するバッチバリアント（例: ingest_records_offset()）があります。バルク取り込みの場合、個別の呼び出しよりもバッチ処理の方が効率的です。
JSON および Protocol Buffers (protobuf) の場合、バッチはアトミックにcommitされます。つまり、バッチ内のすべてのレコードが受け入れられて永続化されるか、バッチ全体が拒否されます。Zerobus Ingest は、これらの形式に対して部分的な upload や部分的な承認を行わないため、テーブルに部分的なバッチが含まれることはありません。検証に失敗したバッチ（スキーマの不一致など）は、一部のレコードを格納して他のレコードを破棄するのではなく、テーブルに触れる前に即座に失敗します。
JSON または protobuf バッチは単一のメッセージとして送信されるため、10 MB の最大メッセージサイズは単一のレコードとバッチ全体の双方に適用されます。つまり、バッチ内のすべてのレコードを合計して 10 MB 以内に収める必要があります。その制限内に収まるようにバッチのサイズを設定してください。レコードサイズを参照してください。
Arrow Flight バッチは例外です
Apache Arrow Flight インジェストは、上記のオールオアナッシングの単一メッセージモデルには従いません。Arrow バッチは JSON や protobuf バッチよりもはるかに大きくなる可能性があり、Arrow Flight パスは大きなバッチを小さな転送メッセージに分割します。これらのメッセージは、1つのアトミックな単位としてではなく、個別に送信および確認されます。その結果：
- JSON および protobuf バッチに適用される 1 メッセージあたり 10 MB の制限は、Arrow バッチには同様には適用されません。大きな Arrow バッチは、サイズを理由に拒否されるのではなく、トランスポートメッセージに分割されます。
- 耐久性はトランスポートメッセージの粒度で確認されるため、非常に大きな論理バッチの場合、途中で障害が発生すると、すべてか無か（all-or-nothing）でcommitされるのではなく、部分的に耐久性を持つ可能性があります。
ingest_batch() 送信したバッチに対して単一の論理オフセットを返し、そのオフセットに対する wait_for_offset() は、バッチを構成するすべてのトランスポートメッセージが確認された後にのみ完了します。完全な Arrow Flight モデル、バッチ処理のガイダンス、および未確認データの回復については、Zerobus Ingest で Arrow Flight を使用するを参照してください。
メッセージをブロックすべきタイミングは？
オフセットでブロッキングを行うと、クライアントコードにおいて、throughputと引き換えに、レコードごとのより強力な耐久性が保証されます。ワークロードに基づいて選択してください：
- ブロックしない : 大容量ストリーミングに適したデフォルト設定です。持続的なスループットが重要であり、集計ベースで（例えば、ストリーム終了時や確認コールバックを通じて）永続性を確認できる場合に適しています。ほとんどのプロデューサーはここから開始してください。
- オフセットでのブロック ：アプリケーションが次の操作を実行する前に、特定のレコードが永続化されたことを確認する必要がある場合に検討してください。例えば：
- データのソース（キューメッセージ、ファイル、アップストリームカーソル）を削除または承認しようとしています。取り込みが失敗した場合にデータを失わないようにする必要があります。
- チェックポイントまたはトランザクション境界でインジェストを行っており、先に進む前に各チェックポイントを耐久性のある状態にする必要があります。
- throughputよりもレコードごとの確認が重要な、少量で価値の高い書き込みを行っています。
ハイスループットのループ内で、すべてのレコードでブロックしないでください。これにより、レコードごとにサーバーへのラウンドトリップでプロデューサーがシリアル化され、throughputが大幅に低下します。代わりに、Databricks では、レコードの大きなチャンクを取り込み、チャンク全体に対して一度だけ耐久性を確認することを推奨しています。これを行うには、最新のオフセットを待機するか、ストリームをフラッシュするという2つの方法があります。個々のレコードごとのブロッキングは、次のアクションの前に単一のレコードを確認する必要がある上記の特定のケースに限定してください。
オフセットを待機する
wait_for_offset() そのオフセットのレコードが永続的に書き込まれたことを Zerobus Ingest が確認するか、タイムアウトするまでブロックします。ストリーム内の特定のポイントを確認するために使用します。通常はチャンクの最後のレコードに使用されます。チャンクを取り込み、ループが返す最終オフセットを保持し、各レコードの後に待機するのではなく、その1つのオフセットで待機します：
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties
sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)
table_properties = TableProperties("main.default.air_quality")
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)
try:
last_offset = 0
for row in records:
last_offset = stream.ingest_record_offset(row)
# Block until everything up to the last record of the chunk is durable
stream.wait_for_offset(last_offset)
print("Chunk durably written.")
finally:
stream.close()
ストリームをフラッシュする
flush() これまでに取り込んだすべてのレコードが永続的に書き込まれるまでブロックし、その後戻ります。wait_for_offset() とは異なり、オフセットを追跡する必要はありません。flush はストリーム上で保留中のすべてが完了するのを待ちます。ストリームは閉じられないため、その後もインジェストを継続できます。
try:
for row in records:
stream.ingest_record_offset(row)
# Block until every pending record is durable
stream.flush()
print("All ingested records durably written.")
finally:
stream.close()
wait_for_offset 対 flush
どちらもチャンクの耐久性を確認します。何を確認するかに基づいて選択してください：
- チェックポイント境界など、特定のレコードまでを確認したい一方で、他のレコードがまだ背後で処理中である可能性がある場合は、
wait_for_offset(offset)を使用します。 - バッチの終了時、アップストリームカーソルを進める前、またはシャットダウン前など、先に進む前に保留中のすべてのレコードが永続化されていることを確認したい場合は、
flush()を使用します。flush()は、構成可能なフラッシュタイムアウトによって制御されます。
close() ストリームをフラッシュして閉じます。これにより、正常なシャットダウン時にレコードが常に永続化されます。必ず finally ブロック内で呼び出してください。
確認応答に非同期で対応する
ブロッキングの代わりに、プロデューサーがフルスピードでプッシュし続ける間、到着する耐久性の確認やエラーに対応したい場合は、ストリームに確認コールバックを登録します。コールバックは、このページのブロッキング呼び出しとは別の機能です。確認コールバックを参照してください。
関連
- 承諾コールバック: 承諾およびエラーに非同期で対応します。
- Zerobus Ingestの使用：クライアントを作成します。
- ストリーム：ストリームとオフセット。
- Zerobus Ingest のエラー処理：エラー処理。