• Projects
  • Service
  • About
  • branding.bz
  • Podcast
  • Tips
  • FAQ
  • Recruit
  • Download
  • Contact
  • branding.bz(ブランド構築SaaS)
  • DESIGN NOW(デザインメディア)
  • X
  • LinkedIn
  • Spotify
  • Facebook

213-0011 神奈川県川崎市高津区久本3-6-7-303

© 2026 ID INC. All rights reserved

claude-skills/スキル
SKILLOfficialdatabase

databricks-zerobus-ingest

プラグイン
databricks
ソース
GitHub で見る ↗
説明

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.

ユースケース
  • Unity Catalogテーブルに直接書き込むプロデューサーを作成するとき
  • Python・Java・Go・TypeScript・Rustで開発するとき
  • Unity CatalogテーブルからProtobufスキーマを自動生成するとき
  • 確認応答と再試行機能を備えたストリーム取り込みを実装するとき
本文(日本語訳)

Zerobus Ingest

Zerobus gRPC API を使用して、データを直接 Databricks Delta テーブルに取り込むクライアントを構築します。

ステータス: 一般提供中。料金は Jobs Serverless SKU に対して請求されます。機能ごとの最新ステータスは Zerobus概要 をご確認ください(ストリーミングテーブルの対象化と Arrow Flight(高速データ転送)などの機能は Beta 版です)。

ドキュメント:

  • Zerobus概要
  • Zerobus Ingest SDK
  • Zerobus制限

Zerobus Ingest とは

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 にはあらかじめインストールされていません:

  • databricks-zerobus-ingest-sdk: 高速ストリーミング取り込み用 Zerobus SDK
  • databricks-sdk: 認証とメタデータ取得用 Databricks ワークスペースクライアント
  • grpcio-tools: Protobuf データ形式のコンパイルに必要(Protobuf のみ)

ノートブックの実行時に pip でインストールするのではなく、ジョブ・クラスタのライブラリ設定 からインストールしてください(以下の「ライブラリをインストールする」セクション参照)。SDK はサーバーレスコンピュートで pip インストールできません。

grpcio-tools と protobuf の互換性

grpcio-tools は実行環境の protobuf バージョンと一致させてください。Protobuf コンパイル時にバージョンエラーが発生した場合、互換性のあるバージョンをピン留めしてください(古い protobuf 5.26/5.29 環境では grpcio-tools==1.62.0 が機能します)。それ以外の場合は最新版を使用してください。


前提条件

以下のすべてのオブジェクトが有効であることを確認してからスキルを実行してください:

  1. Unity Catalog で管理された Delta テーブル — データ取り込み先
  2. サービスプリンシパルの ID とシークレット — ターゲットテーブルに対する変更と読み取り権限が必要
  3. ワークスペース地域用の Zerobus サーバーエンドポイント
  4. 対象言語用 Zerobus Ingest SDK — インストール済み

完全なセットアップ手順は references/1-setup-and-authentication.md をご覧ください。


シンプルな Python 例(JSON)

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 確認応答処理、リトライ、再接続、スループット制限、制約

ワークフロー に必ず従ってください

ワークフロー

  1. 実行計画を表示する
  2. クライアントタイプとデータ形式を決定する — プロトタイプ・シンプルなスキーマは JSON、本番運用・型安全性が必要な場合は Protobuf
  3. スキーマを取得する — Protobuf の場合は references/4-protobuf-schema.md に従い .proto を生成、JSON の場合はレコードのキーがターゲットテーブルのカラムと一致することを確認
  4. Python コードをローカルファイルに記述する — 関連するガイド(例:scripts/zerobus_ingest.py)の手順に従う
  5. ワークスペースにアップロードする: databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts
  6. Databricks で実行する — ジョブまたはノートブック経由で実行
  7. 実行に失敗した場合: ローカルファイルを修正し、再アップロード、再実行

重要

  • SDK はノートブックで実行時に pip インストールするのではなく、ジョブ・クラスタのライブラリ設定からインストールしてください。
  • サーバーレス制限: Zerobus SDK はサーバーレスコンピュートで pip インストールできません。クラシックコンピュートクラスタを使用するか、ノートブックベースの取り込みが必要な場合は Zerobus REST API(Beta 版)を使用してください。
  • 明示的なテーブル権限: サービスプリンシパルがターゲットテーブルに対して明示的な変更・読み取り権限を持つ必要があります。スキーマレベルの継承権限は OAuth フロー(authorization_details)に対して不十分な場合があります。

実行ワークフロー

ステップ 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

実行に失敗した場合:

  1. ジョブ実行ログからエラーを確認する
  2. ローカルの Python ファイルを編集して修正する
  3. 再アップロード: databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts
  4. 再実行: databricks jobs run-now JOB_ID

ライブラリをインストールする

Databricks は 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)  # エポックからのマイクロ秒

キー概念

  • gRPC + Protobuf: Zerobus は gRPC をトランスポートプロトコル(通信規格)として使用します。gRPC で通信でき Protobuf メッセージを構築できるあらゆるアプリケーションが Zerobus に送信できます。
  • JSON または Protobuf シリアル化: JSON はクイックスタート向け、Protobuf は型安全性・前方互換性・パフォーマンス向け。
  • 最低 1 回の配信保証: コネクタは「少なくとも 1 回」の配信を保証します。重複を処理する設計が必要です。
  • 耐久性確認応答: レコードが確実に書き込まれたときに取り込みが確認されます。ingest_record_offset() + wait_for_offset(offset) でオフセット(進捗位置)ベースの追跡、AckCallback で非同期確認、または flush() でバッファされたすべてのレコードが確実に書き込まれるようにしてください。
  • テーブル管理なし: Zerobus はテーブルを作成・変更しません。ターゲットテーブルは事前に作成し、スキーマの進化は自分で管理してください。
  • 単一アベイラビリティゾーン(可用性区間)の耐久性: サービスは 1 つのアベイラビリティゾーンで実行されます。ゾーン障害に備えてください。

よくある問題

問題 解決策
接続が拒否される サーバーエンドポイント形式がクラウド(AWS vs Azure)と一致することを確認。ファイアウォールホワイトリストを確認。
**認証に失敗
原文(English)を表示

Zerobus Ingest

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 Overview
  • Zerobus Ingest SDK
  • Zerobus Limits

What Is Zerobus Ingest?

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


Quick Decision: What Are You Building?

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.


Common Libraries

These libraries are essential for Zerobus data ingestion and are typically NOT pre-installed on Databricks:

  • databricks-zerobus-ingest-sdk: Zerobus SDK for high-performance streaming ingestion
  • databricks-sdk: Databricks workspace client for authentication and metadata
  • grpcio-tools: only needed to compile a .proto for Protobuf serialization

Install 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 and protobuf compatibility

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.


Prerequisites

You must never execute the skill without confirming the below objects are valid:

  1. A Unity Catalog managed Delta table to ingest into
  2. A service principal id and secret with MODIFY and SELECT on the target table
  3. The Zerobus server endpoint for your workspace region
  4. The Zerobus Ingest SDK installed for your target language

See references/1-setup-and-authentication.md for complete setup instructions.


Minimal Python Example (JSON)

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()

Detailed guides

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

Workflow

  1. Display the plan of your execution
  2. Determine the type of client and serialization (JSON for prototypes/simple schemas; Protobuf for production/type safety)
  3. Get schema: for Protobuf, generate the .proto per references/4-protobuf-schema.md; for JSON, ensure record keys match the target table columns
  4. Write Python code to a local file following the instructions in the relevant guide (e.g., scripts/zerobus_ingest.py)
  5. Upload to workspace: databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts
  6. Execute on Databricks using a job or notebook
  7. If execution fails: Edit the local file, re-upload, and re-execute

Important

  • Install the SDK through the job/cluster library configuration, not by pip-installing at runtime in a notebook.
  • Serverless limitation: The Zerobus SDK cannot pip-install on serverless compute. Use classic compute clusters, or use the Zerobus REST API (Beta) for notebook-based ingestion without the SDK.
  • Explicit table grants: Service principals need explicit MODIFY and SELECT grants on the target table. Schema-level inherited permissions may not be sufficient for the authorization_details OAuth flow.

Execution Workflow

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:

  1. Read the error from the job run output
  2. Edit the local Python file to fix the issue
  3. Re-upload: databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts
  4. Re-run: databricks jobs run-now JOB_ID

Installing Libraries

Databricks 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.

Timestamp Format

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

Key Concepts

  • gRPC + Protobuf: Zerobus uses gRPC as its transport protocol. Any application that can communicate via gRPC and construct Protobuf messages can produce to Zerobus.
  • JSON or Protobuf serialization: JSON for quick starts; Protobuf for type safety, forward compatibility, and performance.
  • At-least-once delivery: The connector provides at-least-once guarantees. Design consumers to handle duplicates.
  • Durability ACKs: Ingestion is acknowledged when records are durably written. Use 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.
  • No table management: Zerobus does not create or alter tables. You must pre-create your target table and manage schema evolution yourself.
  • Single-AZ durability: The service runs in a single availability zone. Plan for potential zone outages.

Common Issues

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.

Related Skills

  • databricks-python-sdk - General SDK patterns and WorkspaceClient for table/schema management
  • databricks-pipelines - Downstream pipeline processing of ingested data
  • databricks-unity-catalog - Managing catalogs, schemas, and tables that Zerobus writes to
  • databricks-synthetic-data-gen - Generate test data to feed into Zerobus producers
  • databricks-core - CLI install, profile selection, authentication

Resources

  • Zerobus Overview
  • Zerobus Ingest SDK
  • Zerobus Limits

原文・著作権は Anthropic および各プラグイン作者に帰属します。日本語訳は Claude API による自動翻訳です。