メッセージ遮断と確認応答

Zerobus Ingest SDK には、レコードを取り込むための複数の方法があり、スループットと、得られる耐久性の確認の度合いとの間でトレードオフがあります。このページでは各方法と耐久度をブロックするタイミングを説明しています。確認応答に対してブロックするのではなく非同期的に反応するには、確認応答コールバックを参照してください。

このページの例はPython SDKを使用しています。各メソッドが受け入れる正確なタイムアウトおよび設定オプション(デフォルトやユニットを含む)については、 Zerobus SDKリポジトリをご覧ください。他の言語SDKは同等のオプションを公開しています。

オフセットとは何ですか?

取り込むすべてのレコードにはオフセットが割り当てられています。それはストリーム内での位置です。オフセットとは、特定の記録が耐久性を持って書かれているかどうかを確認したい場合の言及方法です。 Zerobus Ingestは少なくとも一度の配信保証を提供し、オフセットを待つことでクライアントはその保証を特定のレコードで確認できます。

オフセットを確認するということは、レコードが耐久性を持っていることを意味し、まだデルタテーブルでクエリ可能ではないということです。 Zerobus Ingest は、その後まもなく、永続レコードをテーブルに実体化します。レイテンシの数値についてはレイテンシを参照してください。

インジェスト方法

SDKはレコードを取り込むための2つの方法を提供します。 (以下のメソッド名はPython SDKからのものです。他のSDKも同等の手法を公開しています。)

Method	返品ポリシー	次の場合に使用します。
オフセットベースの場合、 `ingest_record_offset()`	レコードがストリームにキューイングされた後の、レコードのオフセット。	推奨デフォルト。レコードを順番にエンキューし、オフセットを待って後で耐久性を確認することもできます。
未来志向、`ingest_record()`	待機可能な `RecordAcknowledgment`。	Deprecated. パフォーマンスが良いならオフセットベースの方が好みです。

オフセットベース(推奨)

ingest_record_offset() レコードを送信し、レコードがストリームにキューに入れられた後にオフセットを返します。呼び出しは呼び出しスレッド上で実行されるため、メソッドを呼び出す順にレコードがキューに入り、返されたオフセットで後で耐久性を wait_for_offset()確認できます。これはほとんどのプロデューサーの推奨デフォルトであり、 Use Zerobus Ingest の例でも使われている方法です。

未来型(非推奨)

ingest_record() は、永続化が完了するのを待機できる RecordAcknowledgment オブジェクトを返します。この方法は、より優れた性能を示すオフセットベース方式に取って代わられました。まだ移行していない既存のコードにのみ使ってください。

レコードごとの取り込みとバッチ取り込みの違い

各インジェスティングメソッドにはバッチバリアント(例: ingest_records_offset())があり、1回の呼び出しでレコードのリストを送信します。大量取り込みでは、個別呼び出しよりもバッチ処理のほうが効率的です。

JSON と Protocol Buffers（protobuf）では、バッチのコミットはアトミックに行われます。つまり、バッチ内のすべてのレコードが受理されて永続化されるか、バッチ全体が拒否されるかのいずれかです。 Zerobus Ingestはこれらのフォーマットに対して部分的なアップロードや部分的な確認応答を行わないため、テーブルには部分的なバッチは含まれません。検証に失敗したバッチ（例えば、スキーマの不一致）は、テーブルに書き込まれる前に即座に失敗し、一部のレコードだけが取り込まれて他が破棄されることはありません。

JSONまたはprotobufバッチは単一のメッセージとして送信されるため、最大10MBのメッセージサイズは1つのレコードとバッチ全体の両方に適用されます。バッチ内のすべてのレコードは合計10MB内に収まらなければなりません。その制限を下回るようにバッチのサイズを決めてください。レコードサイズを参照してください。

Arrow Flightのバッチは例外です

Apache Arrow Flight のインジェスティングは、上記の全か無かの単一メッセージモデルには従いません。 Arrow のバッチは JSON や protobuf のバッチよりもはるかに大きくなることがあり、Arrow Flight のパスでは、大きなバッチを複数の小さなトランスポートメッセージに分割し、それらを単一の不可分な単位としてではなく、個別に送信し、確認応答します。その結果：

JSONやprotobufバッチに適用される10MBのメッセージ制限は、Arrowバッチには同じようには適用されません。大きな Arrow バッチは、サイズ超過を理由に拒否されるのではなく、トランスポートメッセージに分割されます。
永続性はトランスポートメッセージ単位で保証されるため、非常に大きな論理バッチでも、途中で障害が発生した場合は全か無かでコミットされるのではなく、一部のみが永続化されることがあります。

ingest_batch() それでも提出したバッチに対して単一の論理オフセットを返し、そのオフセットの wait_for_offset() はバッチを構成するすべてのトランスポートメッセージが確認された後に完了します。 Arrow Flightモデルの全モデル、バッチングガイダンス、未確認データの復元については、「 Use Arrow Flight with Zerobus Ingest」をご覧ください。

メッセージをブロックすべきタイミングはいつですか?

オフセットでブロッキングすると、スループットが低下する代わりに、クライアントコードにおいてレコード単位でより強固な永続性を保証できます。仕事量に基づいて選びましょう:

ブロックしない:持続的なスループットを重視し、ストリームクロージング時や確認コールバックを通じて集計の耐久性を確認できる高ボリュームストリーミングの正しいデフォルトです。ほとんどのプロデューサーはここから始めるべきです。
オフセットでブロックする:アプリケーションが特定のレコードが耐久性を持つことを知ってから次の行動を取る必要がある場合に考慮してください。例えば：
- データのソース(キューメッセージ、ファイル、上流カーソル)を削除または確認しようとしており、取り込みに失敗しても失わないようにしてはいけません。
- チェックポイント単位またはトランザクション境界ごとに取り込みを行っているため、先に進む前に各チェックポイントが永続化されている必要があります。
- 低ボリュームで高価値の書き込みを行っている場合、レコードごとの確認がスループットよりも重要です。

高スループットのループ処理では、すべてのレコードごとにブロッキングしないでください。これにより、各レコードごとにサーバーへの往復でプロデューサーがシリアライズされ、スループットが大幅に低下します。代わりに、Azure Databricksは大量のレコードを取り込み、その後チャンク全体で耐久性を確認することを推奨しています。それを実現する方法は2つあります。最新のオフセットを待つか、ストリームをフラッシュする方法です。個々のレコードごとのブロッキングは、次の行動前に単一のレコードを確認する必要がある特定のケースに限定すべきです。

オフセットを待つ

wait_for_offset() は、Zerobus Ingest がそのオフセットにあるレコードが永続的に書き込まれたことを確認するか、タイムアウトするまでブロックします。これは、ストリーム内の特定の位置、最も一般的にはチャンクの最後のレコードを確認するために使用します。チャンクを取り込んだら、ループが返す最後のオフセットを保持し、各レコードのたびに待機するのではなく、その最後のオフセットだけを待ちます。

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()とは異なり、オフセットを追跡する必要はありません。フラッシュは、ストリーム上で保留中のすべての処理が完了するまで待機します。ストリームを閉じるわけではないので、その後も摂取を続けられます。

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 vs. フラッシュ

どちらも、しばらくの耐久性が確認されています。確認する内容に基づいて選びましょう:

チェックポイントの境界など特定の記録まで確認したい場合に wait_for_offset(offset) を使い、その背後にまだ他の記録が進行中の場合もあります。
flush()は、先に進む前にすべての保留中のレコードが持続性を持っているか確認したい場合、例えばバッチ終了時、上流カーソルを進める前、シャットダウン前などに使います。 flush() 設定可能なフラッシュタイムアウトによって制御されます。

close() は、ストリームをフラッシュして、閉じるため、レコードは正常なシャットダウン時には必ず永続化されます。必ず finally ブロック内で呼び出してください。

確認応答に非同期的に反応する

ブロックする代わりに、耐久性の確認やエラーに反応したい場合、プロデューサーが全速力でプッシュし続ける間にストリームで確認コールバックを登録してください。コールバックはこのページのブロッキングコールとは別の機能です。確認応答コールバックを参照してください。

確認応答コールバック:確認応答やエラーに対して非同期で対応する。
Zerobus Ingestを使いましょう:クライアントを書く。
ストリーム:ストリームとオフセット。
ゼロバスのインジェストエラー処理:エラー処理。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-19