Zerobus Ingest クライアントを構築して、gRPC(高速な通信プロトコル)経由でほぼリアルタイムのデータを Databricks Delta テーブルに取り込みます。 次のような場合に使用: - メッセージバス(データの流通管理システム)を使わずに Unity Catalog テーブルに直接書き込むプロデューサー(データ送信プログラム)を作成する場合 - Python・Java・Go・TypeScript・Rust の Zerobus Ingest SDK を使って開発する場合 - Unity Catalog テーブルから Protobuf スキーマ(データ構造の定義)を自動生成する場合 - 確認応答と再試行機能を備えたストリーム形式のデータ取り込みを実装する場合
Build Zerobus Ingest clients for near real-time data ingestion into Databricks Delta tables via gRPC. Use when creating producers that write directly to Unity Catalog tables without a message bus, working with the Zerobus Ingest SDK in Python/Java/Go/TypeScript/Rust, generating Protobuf schemas from UC tables, or implementing stream-based ingestion with ACK handling and retry logic.
Zerobus gRPC API を使用して、データを直接 Databricks Delta テーブルに取り込むクライアントを構築します。
ステータス: 一般提供中。料金は Jobs Serverless SKU に対して請求されます。機能ごとの最新ステータスは Zerobus概要 をご確認ください(ストリーミングテーブルの対象化と Arrow Flight(高速データ転送)などの機能は Beta 版です)。
ドキュメント:
Zerobus Ingest は、gRPC を経由して Delta テーブルにレコード単位でデータを直接取り込むサーバーレス(基盤管理不要の)コネクタです。データレイクハウス用のメッセージ基盤(Kafka、Kinesis、Event Hub)を用意する必要がありません。このサービスはスキーマ(データ構造)を検証し、データをターゲットテーブルに書き込み、耐久性の確認(データが安全に保存されたこと)をクライアントに送り返します。
基本的な流れ: SDK 初期化 → ストリーム作成 → レコード取り込み → 確認応答処理 → フラッシュ(バッファ書き込み) → 終了
| 用途 | 言語 | データ形式 | リファレンス |
|---|---|---|---|
| 簡単なプロトタイプ / テスト | Python | JSON | references/2-python-client.md |
| 本番運用の Python プロデューサー | Python | Protobuf(構造化形式) | references/2-python-client.md + references/4-protobuf-schema.md |
| JVM マイクロサービス | Java | Protobuf | references/3-multilanguage-clients.md |
| Go サービス | Go | JSON または Protobuf | references/3-multilanguage-clients.md |
| Node.js / TypeScript アプリ | TypeScript | JSON | references/3-multilanguage-clients.md |
| 高性能システムサービス | Rust | JSON または Protobuf | references/3-multilanguage-clients.md |
| Unity Catalog テーブルからのスキーマ生成 | 任意 | Protobuf | references/4-protobuf-schema.md |
| リトライ / 再接続ロジック | 任意 | 任意 | references/5-operations-and-limits.md |
特に指定がなければ Python をデフォルトにしてください。
これらのライブラリは Zerobus データ取り込みに必須であり、Databricks にはあらかじめインストールされていません:
ノートブックの実行時に pip でインストールするのではなく、ジョブ・クラスタのライブラリ設定 からインストールしてください(以下の「ライブラリをインストールする」セクション参照)。SDK はサーバーレスコンピュートで pip インストールできません。
grpcio-tools は実行環境の protobuf バージョンと一致させてください。Protobuf コンパイル時にバージョンエラーが発生した場合、互換性のあるバージョンをピン留めしてください(古い protobuf 5.26/5.29 環境では grpcio-tools==1.62.0 が機能します)。それ以外の場合は最新版を使用してください。
以下のすべてのオブジェクトが有効であることを確認してからスキルを実行してください:
完全なセットアップ手順は references/1-setup-and-authentication.md をご覧ください。
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import RecordType, StreamConfigurationOptions, TableProperties
sdk = ZerobusSdk(server_endpoint, workspace_url)
options = StreamConfigurationOptions(record_type=RecordType.JSON)
table_props = TableProperties(table_name)
stream = sdk.create_stream(client_id, client_secret, table_props, options)
try:
# JSON ストリームの場合、辞書を渡します。SDK が自動的にシリアル化します
record = {"device_name": "sensor-1", "temp": 22, "humidity": 55}
offset = stream.ingest_record_offset(record)
stream.wait_for_offset(offset) # データが確実に書き込まれるまで待機
finally:
stream.close()
| トピック | ファイル | いつ読むか |
|---|---|---|
| セットアップ・認証 | references/1-setup-and-authentication.md | エンドポイント形式、サービスプリンシパル、SDK インストール |
| Python クライアント | references/2-python-client.md | 同期・非同期 Python、JSON と Protobuf フロー、再利用可能なクライアントクラス |
| 複数言語対応 | references/3-multilanguage-clients.md | Java、Go、TypeScript、Rust SDK の例 |
| Protobuf スキーマ | references/4-protobuf-schema.md | Unity Catalog テーブルから .proto を生成、コンパイル、型マッピング |
| オペレーション・制限 | references/5-operations-and-limits.md | 確認応答処理、リトライ、再接続、スループット制限、制約 |
.proto を生成、JSON の場合はレコードのキーがターゲットテーブルのカラムと一致することを確認scripts/zerobus_ingest.py)の手順に従うdatabricks workspace import-dir ./scripts /Workspace/Users/<user>/scriptsステップ 1: コードをワークスペースにアップロードする
ワークスペースのユーザー名を取得して <user> パスに置き、以下を実行します:
USER=$(databricks current-user me --output json | jq -r .userName)
databricks workspace import-dir ./scripts "/Workspace/Users/$USER/scripts"
ステップ 2: ジョブを作成して実行する
databricks jobs create --json '{
"name": "zerobus-ingest",
"tasks": [{
"task_key": "ingest",
"spark_python_task": {
"python_file": "/Workspace/Users/<user>/scripts/zerobus_ingest.py"
},
"new_cluster": {
"spark_version": "16.1.x-scala2.12",
"node_type_id": "i3.xlarge",
"num_workers": 0
}
}]
}'
databricks jobs run-now JOB_ID
実行に失敗した場合:
databricks workspace import-dir ./scripts /Workspace/Users/<user>/scriptsdatabricks jobs run-now JOB_IDDatabricks は Spark、pandas、numpy および一般的なデータライブラリをデフォルトで提供していますが、Zerobus SDK は プリインストールされていません — 常にジョブ・クラスタのライブラリ設定に databricks-zerobus-ingest-sdk を追加してください。インポートエラーが発生した場合にのみ、その他のライブラリを追加してください。
ジョブ設定に追加:
"libraries": [
{"pypi": {"package": "databricks-zerobus-ingest-sdk>=1.0.0"}}
]
またはクラスタ設定で初期化スクリプトを使用してください。
Delta テーブルの TIMESTAMP カラムは Protobuf の int64 (エポック(基準時刻)からのマイクロ秒)にマップされます(references/4-protobuf-schema.md の型マッピング参照)— 文字列ではなく整数を入力してください:
from datetime import datetime, timezone
event_time = int(datetime.now(timezone.utc).timestamp() * 1_000_000) # エポックからのマイクロ秒
ingest_record_offset() + wait_for_offset(offset) でオフセット(進捗位置)ベースの追跡、AckCallback で非同期確認、または flush() でバッファされたすべてのレコードが確実に書き込まれるようにしてください。| 問題 | 解決策 |
|---|---|
| 接続が拒否される | サーバーエンドポイント形式がクラウド(AWS vs Azure)と一致することを確認。ファイアウォールホワイトリストを確認。 |
| **認証に失敗 |
Build clients that ingest data directly into Databricks Delta tables via the Zerobus gRPC API.
Status: Generally Available. Charges are billed against the Jobs Serverless SKU. Check the Zerobus overview for the current status of specific features (some, such as Streaming-table targets and Arrow Flight, may be in Beta).
Documentation:
Zerobus Ingest is a serverless connector that enables direct, record-by-record data ingestion into Delta tables via gRPC. It eliminates the need for message bus infrastructure (Kafka, Kinesis, Event Hub) for lakehouse-bound data. The service validates schemas, materializes data to target tables, and sends durability acknowledgments back to the client.
Core pattern: SDK init -> create stream -> ingest records -> handle ACKs -> flush -> close
| Scenario | Language | Serialization | Reference |
|---|---|---|---|
| Quick prototype / test harness | Python | JSON | references/2-python-client.md |
| Production Python producer | Python | Protobuf | references/2-python-client.md + references/4-protobuf-schema.md |
| JVM microservice | Java | Protobuf | references/3-multilanguage-clients.md |
| Go service | Go | JSON or Protobuf | references/3-multilanguage-clients.md |
| Node.js / TypeScript app | TypeScript | JSON | references/3-multilanguage-clients.md |
| High-performance system service | Rust | JSON or Protobuf | references/3-multilanguage-clients.md |
| Schema generation from UC table | Any | Protobuf | references/4-protobuf-schema.md |
| Retry / reconnection logic | Any | Any | references/5-operations-and-limits.md |
If not specified, default to python.
These libraries are essential for Zerobus data ingestion and are typically NOT pre-installed on Databricks:
.proto for Protobuf serializationInstall them through the job/cluster library configuration (see Installing Libraries below) rather than pip-installing at runtime — the SDK cannot pip-install on serverless compute.
grpcio-tools must match the runtime's protobuf version. If proto compilation fails with a version error, pin a compatible build (for older protobuf 5.26/5.29 runtimes, grpcio-tools==1.62.0 works); otherwise use the latest release.
You must never execute the skill without confirming the below objects are valid:
MODIFY and SELECT on the target tableSee references/1-setup-and-authentication.md for complete setup instructions.
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import RecordType, StreamConfigurationOptions, TableProperties
sdk = ZerobusSdk(server_endpoint, workspace_url)
options = StreamConfigurationOptions(record_type=RecordType.JSON)
table_props = TableProperties(table_name)
stream = sdk.create_stream(client_id, client_secret, table_props, options)
try:
# Pass a dict for JSON streams; the SDK serializes it.
record = {"device_name": "sensor-1", "temp": 22, "humidity": 55}
offset = stream.ingest_record_offset(record)
stream.wait_for_offset(offset) # Block until durably written
finally:
stream.close()
| Topic | File | When to Read |
|---|---|---|
| Setup & Auth | references/1-setup-and-authentication.md | Endpoint formats, service principals, SDK install |
| Python Client | references/2-python-client.md | Sync/async Python, JSON and Protobuf flows, reusable client class |
| Multi-Language | references/3-multilanguage-clients.md | Java, Go, TypeScript, Rust SDK examples |
| Protobuf Schema | references/4-protobuf-schema.md | Generate .proto from UC table, compile, type mappings |
| Operations & Limits | references/5-operations-and-limits.md | ACK handling, retries, reconnection, throughput limits, constraints |
You must always follow all the steps in the Workflow
.proto per references/4-protobuf-schema.md; for JSON, ensure record keys match the target table columnsscripts/zerobus_ingest.py)databricks workspace import-dir ./scripts /Workspace/Users/<user>/scriptsMODIFY and SELECT grants on the target table. Schema-level inherited permissions may not be sufficient for the authorization_details OAuth flow.Step 1: Upload code to workspace
Get your workspace username for the <user> path segment, then upload:
USER=$(databricks current-user me --output json | jq -r .userName)
databricks workspace import-dir ./scripts "/Workspace/Users/$USER/scripts"
Step 2: Create and run a job
databricks jobs create --json '{
"name": "zerobus-ingest",
"tasks": [{
"task_key": "ingest",
"spark_python_task": {
"python_file": "/Workspace/Users/<user>/scripts/zerobus_ingest.py"
},
"new_cluster": {
"spark_version": "16.1.x-scala2.12",
"node_type_id": "i3.xlarge",
"num_workers": 0
}
}]
}'
databricks jobs run-now JOB_ID
If execution fails:
databricks workspace import-dir ./scripts /Workspace/Users/<user>/scriptsdatabricks jobs run-now JOB_IDDatabricks provides Spark, pandas, numpy, and common data libraries by default, but the Zerobus SDK is never pre-installed — always add databricks-zerobus-ingest-sdk to the job/cluster library config. Only add other libraries if you hit an import error.
Add to the job configuration:
"libraries": [
{"pypi": {"package": "databricks-zerobus-ingest-sdk>=1.0.0"}}
]
Or use init scripts in the cluster configuration.
A Delta TIMESTAMP column maps to a Protobuf int64 of epoch microseconds (see the type mappings in references/4-protobuf-schema.md) — supply an integer, not a string:
from datetime import datetime, timezone
event_time = int(datetime.now(timezone.utc).timestamp() * 1_000_000) # epoch microseconds
ingest_record_offset() + wait_for_offset(offset) for offset-based tracking, an AckCallback for asynchronous confirmation, or flush() to ensure all buffered records are durably written.| Issue | Solution |
|---|---|
| Connection refused | Verify server endpoint format matches your cloud (AWS vs Azure). Check firewall allowlists. |
| Authentication failed | Confirm service principal client_id/secret. Verify GRANT statements on the target table. |
| Schema mismatch | Ensure record fields match the target table schema exactly. Regenerate .proto if table changed. |
| Stream closed unexpectedly | Implement retry with exponential backoff and stream reinitialization. See references/5-operations-and-limits.md. |
| Throughput limits hit | Max 100 MB/s and 15,000 rows/s per stream. Open multiple streams or contact Databricks. |
| Region not supported | Check supported regions in references/5-operations-and-limits.md. |
| Table not found | Ensure table is a managed Delta table in a supported region with correct three-part name. |
| SDK install fails on serverless | The Zerobus SDK cannot be pip-installed on serverless compute. Use classic compute clusters or the REST API (Beta) from notebooks. |
| Error 4024 / authorization_details | Service principal lacks explicit table-level grants. Grant MODIFY and SELECT directly on the target table — schema-level inherited grants may be insufficient. |
原文・著作権は Anthropic および各プラグイン作者に帰属します。日本語訳は Claude API による自動翻訳です。