databricks-zerobus-ingest

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

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 gRPC API直接将数据摄入Databricks Delta表的客户端。

状态: 正式可用。费用将按Jobs Serverless SKU计费。请查看Zerobus概述了解特定功能的当前状态（部分功能如Streaming-table targets和Arrow Flight可能处于Beta阶段）。

文档:

What Is Zerobus Ingest?

什么是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

Zerobus Ingest是一款无服务器连接器，可通过gRPC将数据逐条直接摄入Delta表。它消除了湖仓数据对消息总线基础设施（如Kafka、Kinesis、Event Hub）的依赖。该服务会验证Schema，将数据物化到目标表，并向客户端发送持久性确认信息。

核心流程: SDK初始化 -> 创建流 -> 摄入记录 -> 处理ACK -> 刷新 -> 关闭

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.

场景	开发语言	序列化方式	参考文档
快速原型/测试工具	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
从UC表生成Schema	任意	Protobuf	references/4-protobuf-schema.md
重试/重连逻辑	任意	任意	references/5-operations-and-limits.md

未指定时，默认使用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.

以下是Zerobus数据摄入必不可少的库，通常不会预先安装在Databricks中：

databricks-zerobus-ingest-sdk: 用于高性能流式摄入的Zerobus SDK
databricks-sdk: 用于身份验证和元数据操作的Databricks工作区客户端
grpcio-tools: 仅在编译
```
.proto
```
文件进行Protobuf序列化时需要

请通过作业/集群库配置（见下文的安装库）安装这些库，而非在运行时通过pip安装——SDK无法在无服务器计算环境中通过pip安装。

grpcio-tools and protobuf compatibility

grpcio-tools与protobuf兼容性

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.

grpcio-tools

版本必须与运行环境的

protobuf

版本匹配。如果Proto编译因版本错误失败，请固定兼容版本（对于旧版

protobuf

5.26/5.29运行环境，

grpcio-tools==1.62.0

可行）；否则使用最新版本。

Prerequisites

前提条件

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

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

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

在执行该技能前，必须确认以下对象有效：

用于数据摄入的Unity Catalog托管Delta表
拥有目标表
MODIFY
和
SELECT
权限的服务主体ID及密钥
对应工作区区域的Zerobus服务器端点
针对目标语言安装的Zerobus Ingest SDK

完整设置说明请查看references/1-setup-and-authentication.md。

Minimal Python Example (JSON)

最简Python示例（JSON）

python

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

python

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

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

主题	文件	阅读时机
设置与认证	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 Schema	references/4-protobuf-schema.md	从UC表生成.proto文件、编译、类型映射
操作与限制	references/5-operations-and-limits.md	ACK处理、重试、重连、吞吐量限制、约束条件

必须严格遵循工作流中的所有步骤

Workflow

工作流

Display the plan of your execution
Determine the type of client and serialization (JSON for prototypes/simple schemas; Protobuf for production/type safety)
Get schema: for Protobuf, generate the
```
.proto
```
per references/4-protobuf-schema.md; for JSON, ensure record keys match the target table columns
Write Python code to a local file following the instructions in the relevant guide (e.g.,
```
scripts/zerobus_ingest.py
```
)

Upload to workspace:

databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts

Execute on Databricks using a job or notebook
If execution fails: Edit the local file, re-upload, and re-execute

展示执行计划
确定客户端类型和序列化方式（原型/简单Schema用JSON；生产环境/类型安全用Protobuf）
获取Schema：Protobuf需按照references/4-protobuf-schema.md生成
```
.proto
```
文件；JSON需确保记录键与目标表列名匹配
将Python代码写入本地文件（如
```
scripts/zerobus_ingest.py
```
）

上传至工作区：

databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts

在Databricks上执行：通过作业或笔记本运行
若执行失败：编辑本地文件，重新上传并执行

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.

通过作业/集群库配置安装SDK，而非在笔记本运行时通过pip安装。
无服务器限制：Zerobus SDK无法在无服务器计算环境中通过pip安装。请使用经典计算集群，或使用Zerobus REST API（Beta版）在笔记本中无需SDK即可进行数据摄入。
显式表权限：服务主体需要目标表的显式
```
MODIFY
```
和
```
SELECT
```
权限。Schema级别的继承权限可能无法满足
```
authorization_details
```
OAuth流程的要求。

Execution Workflow

执行工作流

Step 1: Upload code to workspace

Get your workspace username for the

<user>

path segment, then upload:

bash

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

bash

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:

Read the error from the job run output
Edit the local Python file to fix the issue

Re-upload:

databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts

Re-run:
```
databricks jobs run-now JOB_ID
```

步骤1：上传代码至工作区

获取工作区用户名作为

<user>

路径段，然后执行上传：

bash

USER=$(databricks current-user me --output json | jq -r .userName)
databricks workspace import-dir ./scripts "/Workspace/Users/$USER/scripts"

步骤2：创建并运行作业

bash

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

若执行失败：

从作业运行输出中读取错误信息
编辑本地Python文件修复问题

重新上传：

databricks workspace import-dir ./scripts /Workspace/Users/<user>/scripts

重新运行：
```
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:

json

"libraries": [
  {"pypi": {"package": "databricks-zerobus-ingest-sdk>=1.0.0"}}
]

Or use init scripts in the cluster configuration.

Databricks默认提供Spark、pandas、numpy及常见数据库，但Zerobus SDK绝不会预先安装——务必将

databricks-zerobus-ingest-sdk

添加到作业/集群库配置中。仅当遇到导入错误时才添加其他库。

添加到作业配置：

json

"libraries": [
  {"pypi": {"package": "databricks-zerobus-ingest-sdk>=1.0.0"}}
]

或在集群配置中使用初始化脚本。

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:

python

from datetime import datetime, timezone

event_time = int(datetime.now(timezone.utc).timestamp() * 1_000_000)  # epoch microseconds

Delta的

TIMESTAMP

列对应Protobuf的

int64

类型，存储微秒级时间戳（详见references/4-protobuf-schema.md中的类型映射）——需传入整数而非字符串：

python

from datetime import datetime, timezone

event_time = int(datetime.now(timezone.utc).timestamp() * 1_000_000)  # 微秒级时间戳

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.

gRPC + Protobuf: Zerobus采用gRPC作为传输协议。任何可通过gRPC通信并构造Protobuf消息的应用都可向Zerobus生产数据。
JSON或Protobuf序列化: JSON适用于快速启动；Protobuf适用于类型安全、向前兼容和高性能场景。
至少一次投递: 连接器提供至少一次投递保证。需设计消费者以处理重复数据。
持久性ACK: 当记录被持久化写入时会确认摄入状态。可使用
```
ingest_record_offset()
```
+
```
wait_for_offset(offset)
```
进行基于偏移量的跟踪，使用
```
AckCallback
```
进行异步确认，或使用
```
flush()
```
确保所有缓冲记录被持久化写入。
无表管理: Zerobus不创建或修改表。必须预先创建目标表并自行管理Schema演进。
单AZ持久性: 服务在单个可用区运行。需规划应对潜在的可用区故障。

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.

问题	解决方案
连接被拒绝	验证服务器端点格式是否与你的云环境（AWS vs Azure）匹配。检查防火墙允许列表。
认证失败	确认服务主体的client_id/密钥正确。验证目标表上的GRANT语句。
Schema不匹配	确保记录字段与目标表Schema完全匹配。若表结构变更，重新生成.proto文件。
流意外关闭	实现带指数退避的重试和流重新初始化逻辑。详见references/5-operations-and-limits.md。
达到吞吐量限制	每个流最大支持100 MB/s和15,000行/s。可开启多个流或联系Databricks。
区域不支持	在references/5-operations-and-limits.md中查看支持的区域。
表未找到	确保表是受支持区域中的托管Delta表，且三部分名称正确。
SDK在无服务器环境安装失败	Zerobus SDK无法在无服务器计算环境中通过pip安装。请使用经典计算集群或从笔记本调用REST API（Beta版）。
错误4024 / authorization_details	服务主体缺少表级显式权限。直接为目标表授予 `MODIFY` 和 `SELECT` 权限——Schema级的继承权限可能不足。

databricks-zerobus-ingest

Original

Translation

Zerobus Ingest

Zerobus Ingest

What Is Zerobus Ingest?

什么是Zerobus Ingest？

Quick Decision: What Are You Building?

快速决策：你要构建什么？

Common Libraries

常用库

grpcio-tools and protobuf compatibility

grpcio-tools与protobuf兼容性

Prerequisites

前提条件

Minimal Python Example (JSON)

最简Python示例（JSON）

Detailed guides

详细指南

Workflow

工作流

Important

重要提示

Execution Workflow

执行工作流

Installing Libraries

安装库

Timestamp Format

时间戳格式

Key Concepts

核心概念

Common Issues

常见问题

Related Skills

相关技能

Resources

资源