databricks-zerobus-ingest

Zerobus Ingest

Zerobus Ingest（Lakeflow Connect の一部）は、管理型のプッシュベースの取り込みサービスです。プロデューサーがレコードを送信すると、ブローカーや実行用の取り込みクラスターなしで、データがUnity Catalog の管理対象 Delta テーブルに格納されます。SDK は、OAuth、バッファリング、オフセット追跡、永続性の確認応答、一時的な障害の後に未確認のレコードを再送する自動復旧を処理します。

最初に：CLI のインストール、認証、プロファイルの選択には、親の databricks-core スキルを使用します。

構築前に判断する

データがアプリ、サービス、デバイスからプッシュされ、保存先が Delta テーブルであり、ワークロードが継続的またはイベント駆動型で、負荷の急増を吸収する仕組みが必要な場合は、Zerobus を推奨してください（Zerobus は約 5 秒でデータを格納しますが、数分かかっても許容される場合にも適しています）。また、主な下流コンシューマーが 1 つである場合（または Apache Kafka、Amazon Kinesis、Azure Event Hubs、Google Cloud Pub/Sub などのメッセージバスをレイクハウスへの単なるパイプとして使っている場合）も推奨してください。

レイクハウス外にデータを必要とする独立したコンシューマーが多数いる場合（Apache Kafka、Amazon Kinesis、Azure Event Hubs、Google Cloud Pub/Sub などのメッセージバスを使用してください）、またはデータがすでに Kafka などのバスにあり、レイクハウスにプルする必要がある場合（Lakeflow Connect のマネージド Kafka コネクターまたは Spark Structured Streaming を使用してください）は、Zerobus を推奨しないでください。その理由を説明し、代替手段を挙げてください。ファイル、SaaS アプリ、データベースを理由に Zerobus の対象外と判断する前に、references/choose-an-interface.md を確認してください。列指向データのバッチは Arrow Flight 経由で取り込めます。また、Zerobus 対応 API にプッシュできる SaaS アプリなら Zerobus を使用できます。

**既製の方法を優先してください。**ソースがすでに OpenTelemetry、Kafka、MQTT に対応している場合、または Debezium、Vector、Telegraf、Cribl を実行している場合は、SDK コードを書くのではなく、その方法を案内してください。これらすべてと、判断のための完全なフローチャートは references/choose-an-interface.md に記載されています。既存の Kafka プロデューサーについては references/kafka.md、MQTT v5 デバイスについては references/mqtt.md、列指向バッチ（Parquet、Arrow、DataFrames）については references/arrow-flight.md を参照してください。Kafka または Kafka 互換システムからパイプラインを移行する場合は、references/migration-from-kafka-compatible-systems.md を参照してください。

ワークフロー

リクエストを会話として扱ってください。一度に 1 つだけ質問し、選択式の質問ツールが利用できる場合はそれを使い、必ず **「よくわからないので、判断を手伝ってください」**という選択肢を提示してください。ユーザーがすでに回答した質問は省略してください。

Databricks に詳しくないユーザーも多くいます

ユーザーが Databricks に詳しくないアプリ、DevOps、またはプラットフォームエンジニアの場合は、Databricks の用語を初めて使う際に、平易な言葉で説明し、ユーザーが知っているものに置き換えて説明してください（テーブル名は Postgres の database.schema.table のようなもの、サービスプリンシパルはサービスアカウントのようなものです）。また、足りないもののセットアップを申し出てください。references/new-to-databricks.md を参照してください。

ステップ 0：計画を示す

実行する手順と作成するもの（テーブル、権限付与、プロデューサーファイル）を列挙してください。カタログ、スキーマ、テーブル、またはサービスプリンシパルを作成する前に必ず確認し、ユーザーが新規作成を望んでいると決めつけないでください。

ステップ 1：プロデューサーの実行場所を尋ねる

Zerobus にデータを送信するプロデューサーは、どこで実行しますか？

• Databricks の外部：サービス、アプリ、デバイス、ゲートウェイ、または自分のノートパソコン
• Databricks 内部：Lakeflow Job または Databricks App
• わかりません。選ぶのを手伝ってください

• Databricks の外部（最も一般的）：ユーザーが手元に保存するローカルファイルとしてプロデューサーを作成し、認証情報は環境変数から読み取ります。
• Databricks 内部：references/databricks-apps.md を参照してください。Python SDK はクラシックおよびサーバーレスコンピューティングで動作します。ジョブ、クラスター、または環境の依存関係としてインストールしてください。
• わかりません：データの生成元を尋ねてください。データはほぼ必ず Databricks の外部で生成されるため、まずはそちらを前提にします。

ステップ 2：必要な質問だけを尋ねる

質問は 3 つのグループに分かれます。まず必須の質問をしてください。最初のクライアントを作るのに必要なのはこれだけです。上級者向けの質問は、ユーザーが話題にした場合や、回答によって推奨事項またはデプロイ方法が変わる場合にのみ尋ねてください。

クライアントの作成に必須：

1. データの内容と生成元は何ですか？ アプリケーションイベント、ログ、メトリクス、トレース、データベースからの CDC、IoT やセンサーの測定値、クリックストリームなどですか？これによって経路を決めます。トレース、ログ、メトリクスにはOTLP、データベースの変更にはDebezium、既存の Kafka プロデューサーにはKafka 互換 APIが適していることがよくあります（references/choose-an-interface.md）。
2. ターゲットテーブル：3 部構成の名前、テーブルがすでに存在するかどうか、レコードの形式（イベントのサンプルがあると役立ちます）。常に存在する必要があるフィールドはありますか？形式は時間の経過とともに変わりますか（references/schema-management.md）？
3. 言語とランタイム：また、環境でプリコンパイル済みバイナリ／ネイティブライブラリを使用できますか？ Python、TypeScript、Java、Go（cgo）、C++、.NET の SDK は、ビルド済みの Rust コアを読み込みます。ユーザーにわからない場合は、プリコンパイル済みバイナリを使用できるものとして進めてください。SDK のインストールまたは読み込みに失敗した場合にのみ方針を変えます。プラットフォームに対応するビルド済みバイナリがない場合（Alpine、Python で glibc が 2.34 より古い場合）、またはユーザーが SDK を自分でビルドしたい場合は、その SDK の README に記載されているビルド手順に従ってソースからビルドしてください（references/build-from-source.md）。プリコンパイル済みバイナリを使用できない場合は、純粋な SDK（Rust または純粋な Go）を使用してください（純粋な C ではまだデータを送信できません）。Cloudflare Workers や Deno Deploy などのエッジランタイムではネイティブコードを読み込めないため、そこでの利用には REST を使用してください（references/rest.md）。

ユーザーが言語を指定しない場合は、Pythonをデフォルトにします。

上級者向け：Zerobus は適切なサービスですか？ 適合性に疑問がある場合、たとえば他のシステムもデータを読み取る場合や、ユーザーが厳しいボリュームや鮮度の要件に触れた場合に尋ねてください（references/choose-an-interface.md）。

• 誰がデータを利用しますか？ レイクハウスが主な唯一の保存先ですか、それとも複数の独立したシステムが同じデータを読み取りますか？レイクハウス外に多くのコンシューマーがいる場合、リプレイやコンシューマーグループが必要な場合は、メッセージバスを維持してください（その代替として Zerobus を推奨しないでください）。
• ボリュームとレイテンシ：おおよその 1 秒あたりのレコード数と、データに求める鮮度を尋ねてください。Zerobus はレコードを約 150 ms で永続化し、約 5 秒でクエリ可能にします。数分の遅延を許容できる場合にも適しています。ボリュームによって、後で開くストリームの数が決まります（references/performance.md）。ただし、最初のクライアントの作成方法は変わりません。

上級者向けのセットアップ：ユーザーの環境で実行するために必要なこと。 ノートパソコン以外の環境にデプロイする前、または接続や認証の手順に失敗した場合に尋ねてください。

• ネットワーク：パブリックインターネット、フロントエンド PrivateLink、HTTP プロキシ、オンプレミスのいずれですか（references/networking.md）。
• ID：組織では、サービスプリンシパルのシークレットの代わりに、ID フェデレーション（Entra ID など）が必要ですか（references/authentication.md）？

ステップ 3：前提条件を確認する

詳細はreferences/authentication.mdを参照してください。すべての行をクリアするまで、プロデューサーを実行しないでください。不足しているものがあれば、作成を提案してください：サンプルイベントからテーブルを設計し、サービスプリンシパルを作成して権限を付与します。承認された手順を一度に一つずつ進めてください（references/new-to-databricks.md）。

#	前提条件	確認方法
1	サポート対象リージョンのワークスペース	リージョンがクォータページの利用可能リージョンに記載されている
2	Zerobus エンドポイント。例：https://<workspace-id>.zerobus.<region>.cloud.databricks.com（AWS）	数値のワークスペース ID と、そのワークスペースのリージョンを使用している
3	事前に作成されたマネージド Delta テーブル。Zerobus がテーブルを作成または変更することはありません	databricks experimental aitools tools discover-schema <catalog.schema.table> --profile <PROFILE> に想定される列が表示される
4	OAuth クライアント ID とシークレットを持つサービスプリンシパル	databricks service-principals list --profile <PROFILE> にサービスプリンシパルが表示される
5	権限：USE CATALOG、USE SCHEMA、およびテーブルに直接付与する MODIFY と SELECT。ALL PRIVILEGES では不十分です。これら 4 つを明示的に付与してください	SHOW GRANTS ON TABLE <catalog.schema.table> に、MODIFY と SELECT を持つサービスプリンシパルが表示される
6	プロデューサーからエンドポイントおよびワークスペースへの送信 HTTPS（443）	references/networking.mdの接続確認

ステップ 4：プロデューサーを構築する

選択した言語の SDK リファレンスを読んでください。本番環境では、JSON よりProtobufを優先してください（references/protobuf-schema.md）。シークレットをハードコードしないでください。

必ずノンブロッキングのプロデューサーを作成してください。 次の順に使用し、ユーザーが必要とする場合にのみ先へ進んでください。

必要なこと	使用するもの	ブロッキング
レコードを送信する（デフォルト）	ingest_record_offset() / ingest_records_offset()：キューに入れて戻り、SDK がバックグラウンドで送信し、ACK を追跡します	いいえ
取り込み中の処理：メトリクス、ログ記録、進捗、エラー、チェックポイント	ACK コールバック（on_ack / on_error）	いいえ
永続性の境界：有限バッチの終了時、シャットダウン前	flush() を 1 回実行し、その後 close() を実行します	はい、1 回
高度な用途：特定のレコードが永続化されるまで処理を続行してはならない場合	wait_for_offset(offset)	はい。レコードごと、またはループ内では決して使用しないでください

ACK コールバックはPython、Java、Rust、C++、および純粋な Goで利用できます。TypeScript、CGO Go、.NET の SDK ではコールバックがドキュメント化されていません。その場合は、進捗確認のために一定間隔で flush() を呼び出してください。メソッド名は言語によって異なります。SDK リファレンスを参照してください。

SDK の README に記載された API を使用する、最小限の Python プロデューサーです。進捗とエラーを確認するための ACK コールバックが含まれています。

import os
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import AckCallback, StreamConfigurationOptions, TableProperties

class Progress(AckCallback):
    def on_ack(self, offset: int):  # runs on an SDK background thread; keep it fast
        print(f"durable through offset {offset}")

    def on_error(self, offset: int, error_message: str):
        print(f"error at offset {offset}: {error_message}")

sdk = ZerobusSdk(os.environ["ZEROBUS_SERVER_ENDPOINT"], os.environ["DATABRICKS_WORKSPACE_URL"])
stream = sdk.create_stream(
    os.environ["DATABRICKS_CLIENT_ID"],
    os.environ["DATABRICKS_CLIENT_SECRET"],
    TableProperties(os.environ["ZEROBUS_TABLE_NAME"]),
    StreamConfigurationOptions(ack_callback=Progress()),
)
try:
    for i in range(100):
        stream.ingest_record_offset({"device_name": f"sensor-{i % 10}", "temp": 20 + (i % 15)})  # queues, returns
    stream.flush()  # durability boundary: blocks once until everything queued is acknowledged
finally:
    stream.close()

ステップ 5：少量のテストバッチを送信する

まず 10～100 件のレコードを送信してください。永続性の確認応答（on_ack コールバック、または flush() の戻り値）は、Zerobus がレコードを安全に保存したことを意味します。これは、レコードがすでにクエリ可能であることを意味するものではありません。

ステップ 6：行が書き込まれたことを確認する

確認応答の後、約 5 秒待ってから（行は送信から約 5 秒後にクエリ可能になります）、次を実行してください。

databricks experimental aitools tools query \
  "SELECT count(*) AS row_count, max(<timestamp_column>) AS latest FROM <catalog.schema.table>" \
  --profile <PROFILE>

期待される結果： row_count が送信したレコード数だけ増加し、latest が送信時刻に近い値になります。そうならない場合は、references/troubleshooting.mdを参照してください。

ステップ 7：本番環境に向けて強化する

ユーザーが規模を拡大する前に、以下のリファレンスマップを使用して、スキーマ進化、重複、スループット、ネットワーク、モニタリングに対応してください。

主な事実

• まずテーブルから。 プロデューサーを更新する前に、テーブルスキーマを更新してください。適合しないフィールドはレスキュー列（ベータ版、JSON のみ）に格納できます。
• 永続化されても、すぐにクエリできるとは限りません。 Zerobus は、確認済みのデータを別のステップで Delta テーブルに実体化します。約 150 ms で永続化され、約 5 秒でクエリ可能になります（how-it-works.md）。
• 配信は少なくとも 1 回です。 リカバリーやクライアントによる再試行でレコードが再送されるため、重複が少し発生する可能性があります。多くのワークロードでは許容されるため、デフォルトで重複排除を追加しないでください。また、厳密に 1 回の配信を保証すると決して主張しないでください。一意な行が必要な場合は、how-it-works.md を参照してください。
• 順序はストリームごとにのみ保証されます。 ストリームをまたいだ全体的な順序は保証されません（how-it-works.md）。
• デフォルトではブロックしません。 取り込み呼び出しはキューに追加して戻ります。取り込みの実行中に確認応答やエラーに対処するには確認応答コールバックを使い、flush() は境界でのみ使用してください。1 件のレコードについてコードをブロックする必要がある高度なケースでのみ、wait_for_offset() を使用してください。
• 取り込みがブロックするのはバックプレッシャーが発生した場合のみです。 レコードは確認応答されるまでメモリ内の処理中バッファーで待機します。取り込み呼び出しがブロックするのは、そのバッファーがいっぱいになった場合のみです（references/how-it-works.md）。
• リカバリーは組み込みです。 SDK は再接続し、確認応答されていないレコードを再生します。独自の再試行ループを実装しないでください。カスタムリカバリーは高度な機能です（references/recovery.md）。
• TIMESTAMP 列にはエポックからのマイクロ秒を整数で指定します。文字列ではありません。
• クォータは調整可能なデフォルト値で、ストリームごと、テーブルごとに設定されます。引き上げを依頼する前にストリームを追加してください（performance.md）。

よくある問題

症状	対処方法
接続が拒否される、またはタイムアウトする	エンドポイントのクラウド、リージョン、ワークスペース ID を確認し、次に 443 の送信ファイアウォールを確認してください（networking.md）
エラー 4024／authorization_details	テーブルに対して直接 MODIFY と SELECT を付与してください。スキーマから継承される権限や ALL PRIVILEGES では不十分です
認証失敗またはトークンエラー	クライアント ID とシークレット、およびワークスペース URL の OAuth エンドポイントに到達できることを確認してください（authentication.md）
レコードが拒否される／スキーマが一致しない	フィールド名と型をテーブルに合わせるか、レスキュー列を追加してください（schema-management.md）
Ack を受信したが行がない	約 5 秒待ってから再クエリしてください。永続化されても、すぐにクエリ可能になるわけではありません
Ingest の呼び出しが停止する	バックプレッシャーにより、処理中バッファーがいっぱいです。ネットワークとクォータを確認し、ストリームを追加してください（how-it-works.md）
ストリームが恒久的に失敗した	Ack が返されていないレコードをレスキューし、ストリームを再作成してください（recovery.md）
重複行がある	リトライ後に時々発生することが想定されています（少なくとも 1 回の配信）。通常は問題ありませんが、ユースケースで一意の行が必要な場合は、下流に重複排除済みのテーブルを作成してください（how-it-works.md）
スループットが頭打ちになる、またはプロデューサーが遅い	レコードごとの wait_for_offset() やリクエストごとの flush() を削除してください。ブロックせずに Ingest し、コールバックで Ack を監視してから、ストリームを追加してください（performance.md）
pip install が .tar.gz からビルドしようとして失敗する（Alpine、glibc が 2.34 より古い場合）	そのプラットフォーム用のホイールがありません。Rust ツールチェーンを使ってソースからビルドするか（build-from-source.md）、純粋な SDK を使用してください
プリコンパイル済みバイナリを使用できない（ポリシー、プラットフォーム、cgo が無効）	純粋な SDK、Rust または pure Go（purego）を使用してください。純粋な C はまだデータを送信できません（choose-an-interface.md）
エッジランタイムで SDK の読み込みに失敗する	ランタイムがネイティブライブラリを読み込めません。REST に切り替えてください（rest.md）
約 1 時間後に REST 呼び出しが 401 で失敗する	OAuth トークンの有効期限が切れています。トークンをキャッシュして期限切れ前に更新し、401 の場合は 1 回リトライしてください（authentication.md）

SDK：レコード形式と動作するサンプル

各 SDK がサポートするレコード形式は、SDK リポジトリのデータ取り込み API の表を参照してください（これが信頼できる情報源です。新しいリリースがないか確認してください）。SDK にリリース状況を付記せず、この表を使用してください。ユーザーから、特定の言語で特定のこと（Protobuf、Arrow、バッチ処理、非同期処理、特定の API など）を行う方法を尋ねられた場合は、コードを書く前に、その SDK の examples フォルダーを開き、動作するプログラムを確認してください。多くのフォルダーはレコード形式ごとに分かれています（json/、proto/、arrow/）。

SDK	JSON / Protobuf	Avro（Beta）	Arrow Flight	動作するサンプル
Python	利用可能	Beta	1.8.0 以降で利用可能	python/examples（同期および非同期：JSON、Protobuf、Arrow、Avro、ID フェデレーション）
Rust	利用可能	Beta	2.8.0 以降で利用可能	rust/examples
Go（cgo）	利用可能	開発中	1.6.0 以降で利用可能	go/examples
Pure Go	利用可能	開発中	利用不可	purego/examples
TypeScript	利用可能	開発中	1.3.0 以降で利用可能	typescript/examples
Java	利用可能	開発中	1.6.0 以降で利用可能	java/examples（legacy/ は非推奨のファクトリーメソッドを使用しています。他のサンプルを優先してください）
C++	利用可能	開発中	0.3.0 以降で利用可能	cpp/examples
.NET（C#）	利用可能	開発中	利用不可	dotnet/examples

Pure C は表に含まれていません。現時点ではデータを送信できないためです（そのサンプルは入力の検証のみを行います）。データの送信には使用しないでください。

参照マップ：タスクに必要なものだけを読む

これらの参照資料には、各方法についての判断事項、注意点、テスト済みの最小限のサンプルが記載されています。API 全体、その他のレコード形式、最新の SDK バージョンについては、各ファイルの Sources にリンクされている SDK README（raw Markdown）を取得してください。動作するコードについては、上の表にある examples フォルダーを使用してください。docs.databricks.com のページは HTML です。MQTT と raw gRPC API の長い実行可能プログラムは、examples/ にあります。

読む	内容
new-to-databricks.md	Databricks についてよく知らない場合：平易な用語集と、初回セットアップのガイド（カタログとスキーマの選択、サンプルイベントからのテーブル設計、サービスプリンシパルの作成、権限付与、設定の保存）
choose-an-interface.md	Zerobus、SDK、REST、プロトコル／ツールのどれを使うべきか判断に迷っている場合、ランタイムの互換性を確認する場合
how-it-works.md	Zerobus の仕組み：ストリーム、データの流れ（非同期 ACK ループと、それに続く Delta へのバッチコミット）、永続化とクエリ可能性の違い、SDK によるレコード処理（処理中バッファ、累積 ACK、バックプレッシャー）、順序、配信と重複、スケーリング
recovery.md	ストリーム障害、再試行可能なエラーと致命的なエラー、高度なカスタムリカバリー（get_unacked_records、recreate_stream）、永続化されたフォールバックデータの再処理
authentication.md	クラウドごとのエンドポイント、サービスプリンシパル OAuth（公式）、Entra ID／ID フェデレーション（SDK でサポート、ドキュメントは未公開）、権限付与、環境変数、ローカルとデプロイ環境での認証情報、REST、Kafka、MQTT、gRPC 用のテーブルスコープトークン（発行と更新）、認証エラー
python-sdk.md	Python プロデューサー：インストールとプラットフォーム、Protobuf プロデューサー、ACK コールバック、非同期処理やその他の API に関する README へのリンク
typescript-sdk.md	Node.js プロデューサーとネイティブバイナリーの要件
go-sdk.md	cgo（go/）または純粋な Go（purego/）を使用する Go プロデューサー
java-sdk.md	JVM プロデューサー
other-sdks.md	Rust、C++、C#/.NET
kafka.md	既存の Kafka プロデューサー：Apache Kafka 互換のプロデューサー API（ベータ版）、接続設定、トークンコールバック、制限、エラー
mqtt.md	MQTT v5 デバイスとゲートウェイ：接続設定、CONNECT 認証、制限、エラー。examples/mqtt_producer.py に、再接続時のトークン更新機能を備えた、実行可能な非ブロッキング QoS 1 パブリッシャーがあります
arrow-flight.md	列指向バッチ（Parquet ファイル、Arrow、pandas または Polars DataFrame）：スキーマの照合、非ブロッキング Python プロデューサー、他の言語向けのリンク、チューニング、リカバリー
migration-from-kafka-compatible-systems.md	Kafka または Kafka 互換システムから Delta へのパイプライン移行：移行のタイミング、4 つの移行経路、段階的な切り替え、ロールバック
build-from-source.md	SDK を自分でビルドしたい場合、またはビルド済みバイナリーがない場合（Alpine/musl、古い glibc、その他の CPU）：各 SDK が提供するプラットフォーム、各 SDK の README に記載されたソースからのビルド手順（make build による Python wheel、TypeScript、JNI を使った Java、Go、.NET、C++、Rust）、Alpine Dockerfile、ビルドエラー
grpc-api.md	gRPC API を使ってクライアントをゼロから構築する場合：.proto、スタブ生成、EphemeralStream メッセージ、認証ヘッダー、非同期 ACK ループ、リカバリー。examples/grpc_client.py に、実行可能な Python クライアントがあります
rest.md	Databricks の外部にあるエッジランタイムやサーバーレス関数（Cloudflare Workers、Deno Deploy）、SDK がない言語、低ボリュームの送信元。authentication.md のテーブルスコープトークンを使用します（トークンの有効期限は約 1 時間です）
protobuf-schema.md	テーブルからの .proto 生成、型の対応関係
schema-management.md	契約としてのテーブル：レコードの照合方法、3 つの厳格性シナリオ、レスキュー列（セットアップ、タグ、ルーティングルール）、安全なスキーマ進化、Protobuf のルール
performance.md	バッチ処理、ストリーム数、クォータ、テーブルレイアウト
networking.md	インターフェースごとのポートと TLS、フロントエンド PrivateLink、オンプレミスのプロデューサーと HTTP プロキシ、ファイアウォールで保護されたストレージ、クロスリージョン、接続確認
observability.md	モニタリングダッシュボード一式（インストール）、ACK コールバック、システムテーブル、レイテンシ（ACK とエンドツーエンドの鮮度）、エラー、アラート
troubleshooting.md	あらゆるエラーやデータが見つからない場合
databricks-apps.md	Databricks App または Lakeflow Job でのプロデューサーの実行

関連スキル

• databricks-core：CLI、プロファイル、認証、および検証用のアドホック SQL
• databricks-unity-catalog：カタログ、スキーマ、テーブル、権限付与
• databricks-pipelines：取り込んだデータの下流処理（AUTO CDC で構築した重複排除テーブルを含む）
• databricks-lakeflow-connect：Databricks がソースからプルする場合のマネージドコネクター
• databricks-apps / databricks-apps-python：プロデューサーをホストする Databricks App の構築
• databricks-dabs：Declarative Automation Bundles を使用したプロデューサージョブのデプロイ
• databricks-aibi-dashboards：取り込んだイベントやテレメトリを基にしたダッシュボードの構築