Skip to main content

Use Zerobus Ingest

This page describes how to ingest data using Zerobus Ingest in Lakeflow Connect.

Get started with Zerobus Ingest

note

If you have a client-side firewall, add the IP address used by Zerobus Ingest to your allowlist. To view IP addresses by region, see IP addresses and domains for Databricks services and assets.

Before you start, confirm Zerobus Ingest is available in your workspace's region. See Ingestion availability.

  1. Get a Zerobus Ingest URL.
  2. Create or identify the table you want to ingest data into.
  3. Create a service principal and grant privileges to the table.
  4. Connect a client or exporter to start sending data.

Choose the guide for your use case:

  • Ingest your own data: Use the Zerobus Ingest SDKs or REST API with a schema you define. Follow the instructions on this page.

  • Ingest OpenTelemetry data: Use standard OpenTelemetry SDKs or collectors to send traces, logs, and metrics into predefined table schemas. For full instructions, see Ingest OpenTelemetry data with Zerobus Ingest.

Choose an interface

Zerobus Ingest supports several interfaces, all writing directly into Unity Catalog Delta tables. In short:

  • SDKs over gRPC: highest sustained throughput, best for high-volume streaming producers.
  • REST: stateless, best for large fleets of lightweight or "chatty" edge devices.
  • OpenTelemetry (OTLP): for systems already emitting OpenTelemetry traces, logs, and metrics. See Ingest OpenTelemetry data with Zerobus Ingest.

For a full comparison and how to choose, see API protocols. Over the SDKs, you can also choose a record format (JSON, Protocol Buffers (protobuf), or Apache Arrow). See Message types. The rest of this page uses the SDKs and the REST API.

Get your workspace URL and Zerobus Ingest endpoint

Your workspace URL appears in the browser when you log in. While the full URL follows the format https://<databricks-instance>.com/o=XXXXX, the workspace URL consists of everything before the /o=XXXXX. For example, given the following full URL, you can determine the workspace URL and workspace ID.

  • Full URL: https://abcd-teste2-test-spcse2.cloud.databricks.com/?o=2281745829657864#
  • Workspace URL: https://abcd-teste2-test-spcse2.cloud.databricks.com
  • Workspace ID: 2281745829657864

The server endpoint depends on the workspace and region:

  • Server endpoint: <workspace-id>.zerobus.<region>.cloud.databricks.com

To find your workspace region, open the workspace switcher in the top navigation bar of the Databricks UI. The region is displayed below each workspace name (for example, us-west-2). You can also find it in the account console under Workspaces.

For region availability, see Zerobus Ingest quotas.

Create or identify the target table

Identify the target table that you want to ingest data into. To create a new target table, run the CREATE TABLE SQL command. For example, create a new table named unity.default.air_quality.

SQL
    CREATE TABLE unity.default.air_quality (
device_name STRING, temp INT, humidity LONG);

Zerobus Ingest can write to both managed Delta tables and streaming tables, which work the same way, with the same limits and quotas.

note

For OpenTelemetry ingestion, tables must use predefined schemas for each signal type (traces, logs, metrics). See Create target tables in Unity Catalog.

Your table schema is the contract for what Zerobus Ingest accepts, and Zerobus Ingest never auto-evolves it. Plan schema changes proactively: evolve the table first, then update producers. Zerobus Ingest writes records that no longer fit after a breaking table change to a durable fallback location instead of dropping them. See Schema management and Recovering data from the durable fallback location.

By default, Zerobus Ingest rejects records with fields that don't match the target table's schema. To capture those fields instead of losing them, configure a rescue column. See Zerobus rescue column.

Create a service principal and grant permissions

A service principal is a specialized identity that provides more security than personalized accounts. For more information about service principals and how to use them for authentication, see Authorize service principal access to Databricks with OAuth.

You can create and manage service principals programmatically with the Databricks REST API or SDKs, or through the workspace UI as described below. The permission grants at the end of this section are SQL you can run from any client.

  1. To create a service principal, go to Settings > Identity and Access.

  2. Under Service principals, select Manage.

  3. Click Add service principal.

  4. In the Add service principal window, create a new service principal by clicking Add new.

  5. Generate and save the client ID and the client secret for the service principal.

  6. Grant the required permissions for the catalog, the schema, and the table to the service principal.

    1. On the Service principal page, go to the Configurations tab.
    2. Copy the Application Id (UUID).
    3. Use the following SQL to grant permissions, replacing the example UUID and catalog, schema name, and table names if required.
    SQL
    GRANT USE CATALOG ON CATALOG <catalog> TO `<UUID>`;
    GRANT USE SCHEMA ON SCHEMA <catalog.schema> TO `<UUID>`;
    GRANT MODIFY, SELECT ON TABLE <catalog.schema.table_name> TO `<UUID>`;

Write a client

Use a Zerobus SDK in your preferred programming language or the REST API to ingest data into your target table. The SDKs are open source. For the full library, language-specific documentation, and additional examples, see the Zerobus SDK repository.

The examples below use ingest_record_offset, which preserves the order in which you send records.

Python 3.9 or higher is required. The SDK provides high throughput and efficient network I/O through an async runtime. It supports JSON (simplest) and Protocol Buffers (recommended for production). The SDK also supports both sync and async implementations, as well as the offset-based and future-based ingestion methods.

Bash
pip install databricks-zerobus-ingest-sdk

JSON example:

Python
import logging
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties

# See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
SERVER_ENDPOINT="https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
DATABRICKS_WORKSPACE_URL="https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
TABLE_NAME="main.default.air_quality"
CLIENT_ID="your-client-id"
CLIENT_SECRET="your-client-secret"

sdk = ZerobusSdk(
SERVER_ENDPOINT,
DATABRICKS_WORKSPACE_URL
)

table_properties = TableProperties(TABLE_NAME)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)

try:
for i in range(1000):
record_dict = {
"device_name": f"sensor-{i}",
"temp": 20 + i % 15,
"humidity": 50 + i % 40
}
stream.ingest_record_offset(record_dict)
finally:
stream.close()

The examples above use the offset-based ingest_record_offset method without waiting on the returned offset. To learn about the available ingestion methods, when to wait for durability confirmation on an offset, and how to track progress with an acknowledgment callback, see Message blocking and acknowledgment.

Protocol Buffers: For type-safe ingestion, pass a protobuf descriptor to TableProperties (the format is selected automatically). Generate a schema from your table using the generate_proto tool, compile it with protoc, then pass the compiled descriptor to create the stream.

Arrow Flight (Beta): For columnar or batch-oriented ingestion of Apache Arrow RecordBatch data over the same gRPC connection, see Use Arrow Flight with Zerobus Ingest. Requires the [arrow] extra: pip install "databricks-zerobus-ingest-sdk[arrow]" pyarrow.

For complete documentation, configuration options, batch ingestion, and Protocol Buffer examples, see the Python SDK repository.

Handle errors

The examples above show the happy path. In production, wrap ingestion in error handling. The SDK retries transient errors, such as network issues, automatically through its built-in recovery. Failures it can't recover from, such as invalid credentials or a missing table, surface as ZerobusException:

Python
from zerobus.sdk.shared import ZerobusException

try:
stream.ingest_record_offset(record)
except ZerobusException as e:
# Handle the failure: log it, fix the cause, recover on a new stream, or stop.
...

The SDKs also recover from transient failures automatically and let you rescue unacknowledged records when a stream fails permanently. For resilient-client patterns and the full error reference, see Recovery and retry patterns and Zerobus Ingest error handling.

Next steps