databricks-zerobus-ingest

Zerobus Ingest

Zerobus Ingest（属于 Lakeflow Connect）是一项托管的推送式数据摄取服务：生产者发送记录，记录便会写入 Unity Catalog 托管的 Delta 表，无需运行代理或数据摄取集群。SDK 会为你完成实际工作：OAuth、缓冲、偏移量跟踪、持久化确认，以及在发生暂时性故障后重放未确认记录的自动恢复。

首先：使用父级 databricks-core skill 安装 CLI、进行身份验证并选择配置文件。

构建前先做决定

当数据由应用、服务或设备推送；目标是 Delta 表；工作负载持续运行或由事件触发，且需要缓冲机制（Zerobus 大约在 5 秒内写入数据；即使能接受数分钟的延迟，它也仍然适用）；并且只有一个主要下游消费者（或者 Apache Kafka、Amazon Kinesis、Azure Event Hubs 或 Google Cloud Pub/Sub 等消息总线只是通往 lakehouse 的管道）时，推荐使用 Zerobus。

当 lakehouse 外有许多相互独立的消费者需要数据时（使用 Apache Kafka、Amazon Kinesis、Azure Event Hubs 或 Google Cloud Pub/Sub 等消息总线），或者数据已在 Kafka 或其他总线中，必须拉取到 lakehouse 时（使用 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。

工作流

将请求视为一段对话：每次只问一个问题，如果有可用的多选题工具，就使用该工具，并始终提供“不确定，请帮我决定”选项。跳过用户已经回答的问题。

许多用户不了解 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 可在经典和 serverless 计算上运行；将其作为作业、集群或环境依赖项安装。
• 不确定：询问数据由什么生成。数据几乎总是源自 Databricks 外部，因此默认选择外部。

第 2 步：只询问仍需了解的问题

问题分为三组。先询问必需的问题：这些信息足以构建第一个客户端。仅当用户提及相关话题，或答案会改变你的建议或部署方案时，才询问高级问题。

构建客户端所需的信息：

1. 数据是什么，来自哪里？ 应用事件、日志、指标、跟踪、数据库 CDC、IoT 或传感器读数、点击流？这决定了接入方式：跟踪、日志和指标通常适合 OTLP，数据库变更适合 Debezium，现有 Kafka 生产端适合 Kafka 兼容 API（references/choose-an-interface.md）。
2. 目标表：三段式名称、表是否已存在，以及记录的结构（示例事件会有所帮助）。是否必须始终包含某些字段？结构会随时间变化吗（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）：

• 谁会使用这些数据？ Lakehouse 是唯一的主要目标，还是多个独立系统会读取相同数据？Lakehouse 外部有许多使用者、需要重放或使用者组时，意味着保留消息总线（不要建议用 Zerobus 替代它）。
• 吞吐量和延迟：大致每秒记录数，以及数据必须达到多新的状态。Zerobus 可在约 150 ms 内持久化记录，并在约 5 秒后支持查询；即使几分钟也能接受，仍然适用。吞吐量决定之后需要打开多少个流（references/performance.md）；不会改变第一个客户端的构建方式。

高级设置：在用户环境中运行所需的条件。 在笔记本电脑以外部署之前，或连接或身份验证步骤失败时询问：

• 网络：公共互联网、前端 PrivateLink、HTTP 代理，还是本地部署环境（references/networking.md）。
• 身份：组织是否要求使用身份联合（例如 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 不足以满足要求；请明确授予这四项权限	SHOW GRANTS ON TABLE <catalog.schema.table> 列出具有 MODIFY 和 SELECT 权限的服务主体
6	从生产者到端点和工作区的出站 HTTPS（443）	参阅 references/networking.md 中的连接检查

第 4 步：构建生产者

阅读所选语言的 SDK 参考。对于生产环境，优先选择 Protobuf，而非 JSON（references/protobuf-schema.md）。切勿将密钥硬编码。

始终编写非阻塞生产者。 使用以下阶梯，仅在用户需要时才继续向下：

需求	使用方式	是否阻塞？
发送记录（默认）	ingest_record_offset() / ingest_records_offset()：将记录加入队列后立即返回；SDK 在后台发送并跟踪确认	否
在数据摄取期间执行操作：指标、日志记录、进度、错误、检查点	确认回调（on_ack / on_error）	否
持久性边界：有界批次结束时、关闭之前	调用一次 flush()，然后调用 close()	是，仅一次
高级：必须等到某条特定记录持久化后才能继续	wait_for_offset(offset)	是。切勿对每条记录调用，也不要在循环中调用

Python、Java、Rust、C++ 和纯 Go 均支持确认回调。TypeScript、CGO Go 和 .NET SDK 未记录此类回调；在这些 SDK 中，请按固定间隔调用 flush() 以查看进度。方法名称因语言而异；请参考 SDK 文档。

最简 Python 生产者（使用 SDK README 中的 API），通过确认回调查看进度和错误：

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 步：为生产环境做好准备

在用户扩大规模之前，请根据下方参考索引，处理架构演进、重复记录、吞吐量、网络和监控问题。

关键事实

• 先更新表。 更新生产方之前，先更新表架构。不适用的字段可以放入补救列（Beta，仅限 JSON）。
• 持久化不代表可查询。 Zerobus 会通过单独的步骤将已确认的数据写入 Delta 表：约 150 ms 后持久化，约 5 秒后可查询（how-it-works.md）。
• 至少一次投递。 恢复或客户端重试可能会重新发送记录，因此可能出现少量重复。大多数工作负载都能接受；默认不要添加去重，也绝不要声称实现了恰好一次投递。如果用户需要唯一行，请参阅 how-it-works.md。
• 顺序仅限单个流内。 不保证不同流之间的全局顺序（how-it-works.md）。
• 默认采用非阻塞模式。 摄取调用会将数据加入队列后立即返回。使用确认回调在摄取运行期间处理确认和错误，仅在边界处调用 flush()；只有在高级场景中，代码必须阻塞等待单条记录时，才使用 wait_for_offset()。
• 仅在背压时阻塞摄取。 记录会在内存中的飞行中缓冲区内等待确认；只有缓冲区已满时，摄取调用才会阻塞（references/how-it-works.md）。
• 内置恢复机制。 SDK 会重新连接并重放未确认的记录。不要自行编写重试循环；自定义恢复属于高级用法（references/recovery.md）。
• TIMESTAMP 列要求使用 epoch 微秒（整数），而非字符串。
• 配额是可调整的默认值，按流和表分别设置。请求提高配额前，请先增加流的数量（performance.md）。

常见问题

症状	解决方法
连接被拒绝或超时	检查端点的云、区域和工作区 ID，然后检查 443 端口上的出站防火墙 (networking.md)
错误 4024 / authorization_details	直接在表上授予 MODIFY 和 SELECT 权限；继承的架构授权和 ALL PRIVILEGES 都不够
身份验证失败或令牌错误	检查客户端 ID 和密钥，并确认工作区 URL 的 OAuth 端点可访问 (authentication.md)
记录被拒绝 / 架构不匹配	确保字段名称和类型与表匹配，或添加一个容错列 (schema-management.md)
收到确认但没有行	等待约 5 秒后重新查询；数据持久化后不一定能立即查询
写入调用卡住	背压：在途缓冲区已满。检查网络和配额，然后增加流的数量 (how-it-works.md)
流永久失败	保存未确认的记录并重新创建流 (recovery.md)
重复行	重试后偶尔出现重复行是正常情况（至少一次投递）。通常没有问题；如果用例需要行唯一，请在下游构建去重表 (how-it-works.md)
吞吐量停滞或生产者速度慢	移除任何逐条记录调用的 wait_for_offset() 或逐请求调用的 flush()；以非阻塞方式写入，并通过回调观察确认，然后增加流的数量 (performance.md)
pip install 从 .tar.gz 构建并失败（Alpine、glibc 早于 2.34）	该平台没有可用的 wheel。使用 Rust 工具链从源代码构建 (build-from-source.md)，或使用纯 SDK
不允许使用预编译二进制文件（策略限制、平台限制、禁用 cgo）	使用纯 SDK：Rust 或 纯 Go（purego）。纯 C 目前尚不能发送数据 (choose-an-interface.md)
SDK 无法在边缘运行时加载	运行时无法加载原生库；请改用 REST (rest.md)
REST 调用在约 1 小时后因 401 失败	OAuth 令牌已过期。缓存令牌并在过期前刷新，遇到 401 时重试一次 (authentication.md)

SDK：记录格式和可运行示例

各 SDK 支持的记录格式见 SDK 仓库中的数据摄取 API 表（权威来源；请查看以了解较新版本）。不要为 SDK 标注发布状态；请使用此表。当用户询问如何用某种语言实现特定功能（Protobuf、Arrow、批处理、异步、某个特定 API）时，编写代码前请先打开该 SDK 的示例文件夹，查找可运行程序。大多数文件夹按记录格式划分（json/、proto/、arrow/）。

SDK	JSON / Protobuf	Avro（Beta）	Arrow Flight	可运行示例
Python	可用	Beta	自 1.8.0 起可用	python/examples（同步和异步：JSON、Protobuf、Arrow、Avro、身份联合）
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（原始 Markdown）；如需可运行代码，请使用上表中的示例文件夹。docs.databricks.com 页面为 HTML。MQTT 和原始 gRPC API 的长篇可运行程序位于 examples/。

阅读	适用场景
new-to-databricks.md	用户不了解 Databricks：通俗易懂的术语表，以及引导式首次设置（选择目录和架构，根据示例事件设计表，创建服务主体，授予权限，保存设置）
choose-an-interface.md	不确定该使用 Zerobus、SDK、REST 还是协议／工具路径；检查运行时兼容性
how-it-works.md	Zerobus 的工作原理：流、数据路径（异步确认循环，然后批量提交到 Delta）、持久化与可查询性的区别、SDK 如何处理记录（在途缓冲区、累积确认、背压）、排序、交付与重复、扩展
recovery.md	流故障、可重试与致命错误、高级自定义恢复（get_unacked_records、recreate_stream）、持久化回退重处理
authentication.md	各云的端点、服务主体 OAuth（官方）、Entra ID／身份联合（SDK 已支持，文档待补充）、权限授予、环境变量、本地与已部署凭据、用于 REST、Kafka、MQTT 和 gRPC 的表级令牌（签发与刷新）、身份验证错误
python-sdk.md	Python 生产者：安装和平台、Protobuf 生产者、确认回调；异步和其他 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（Beta）、连接设置、令牌回调、限制、错误
mqtt.md	MQTT v5 设备和网关：连接设置、CONNECT 身份验证、限制、错误；可运行的非阻塞 QoS 1 发布者，在 examples/mqtt_producer.py 中支持重连时刷新令牌
arrow-flight.md	列式批次（Parquet 文件、Arrow、pandas 或 Polars DataFrame）：架构匹配、非阻塞 Python 生产者、其他语言的链接、调优、恢复
migration-from-kafka-compatible-systems.md	将管道从 Kafka 或兼容 Kafka 的系统迁移到 Delta：何时迁移、四种路径、分步切换、回滚
build-from-source.md	用户想自行构建 SDK，或没有预构建二进制文件（Alpine/musl、较旧的 glibc、其他 CPU）：各 SDK 支持的平台、各 SDK README 中的源码构建步骤（通过 make build 构建 Python wheel、TypeScript、Java with JNI、Go、.NET、C++、Rust）、Alpine Dockerfile、构建错误
grpc-api.md	从头构建客户端以使用 gRPC API：.proto、存根生成、EphemeralStream 消息、身份验证标头、异步确认循环、恢复；可运行的 Python 客户端见 examples/grpc_client.py
rest.md	Databricks 之外的边缘运行时和无服务器函数（Cloudflare Workers、Deno Deploy）、没有 SDK 的语言、低流量发送方。使用来自 authentication.md 的表级令牌（令牌约一小时后过期）
protobuf-schema.md	从表生成 .proto、类型映射
schema-management.md	以表作为契约：记录如何匹配、三种严格性场景、救援列（设置、标记、路由规则）、安全的架构演进、Protobuf 规则
performance.md	批处理、流数量、配额、表布局
networking.md	各接口的端口和 TLS、前端 PrivateLink、本地部署的生产者和 HTTP 代理、受防火墙限制的存储、跨区域、连接检查
observability.md	监控仪表板套件（安装）、确认回调、系统表、延迟（确认延迟和端到端数据新鲜度）、错误、警报
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：基于摄取的事件或遥测数据构建仪表板