Use Zerobus Ingest

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

Get started with Zerobus Ingest

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

Get a Zerobus Ingest URL.
Create or identify the table you want to ingest data into.
Create a service principal and grant privileges to the table.
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.

Kafka-compatible APIs (Beta): for producers that already speak the Kafka protocol. See Use Kafka-compatible APIs 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>.net/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.azuredatabricks.net/?o=2281745829657864#
Workspace URL: https://abcd-teste2-test-spcse2.azuredatabricks.net
Workspace ID: 2281745829657864

The server endpoint depends on the workspace and region:

Server endpoint: <workspace-id>.zerobus.<region>.azuredatabricks.net

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, eastus). 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.

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

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.

Ingest into a streaming table

Important

Ingesting into Streaming tables using Zerobus Ingest is in Beta. Workspace admins can control access to this feature from the Previews page. See Manage Azure Databricks previews.

To create a new streaming table, run the CREATE STREAMING TABLE SQL command. For example:

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

After the streaming table is created, ingest into it using any of the interfaces in Write a client, exactly as you would for a standard Delta table. Writing to a streaming table works the same way as writing to a managed Delta table, with the same limits and quotas.

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 Azure Databricks with OAuth.

You can create and manage service principals programmatically with the Azure 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.

To create a service principal, go to Settings > Identity and Access.
Under Service principals, select Manage.
Click Add service principal.
In the Add service principal window, create a new service principal by clicking Add new.
Generate and save the client ID and the client secret for the service principal.
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.
```
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 SDK

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.

pip install databricks-zerobus-ingest-sdk

JSON example:

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.eastus.azuredatabricks.net"
DATABRICKS_WORKSPACE_URL="https://adb-1234567890123456.12.azuredatabricks.net"
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.

Rust SDK

Rust 1.70 or higher is required. The SDK uses async I/O and gRPC for high-throughput ingestion. It supports JSON (simplest) and Protocol Buffers (recommended for production).

First, import the package.

cargo add databricks-zerobus-ingest-sdk

Or add it to your Cargo.toml.

[dependencies]
databricks-zerobus-ingest-sdk = "2.0.0" # Latest version at time of publication

JSON example:

  use databricks_zerobus_ingest_sdk::{JsonString, ZerobusSdk};
  use std::error::Error;

  // See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
  const DATABRICKS_WORKSPACE_URL: &str = "https://adb-1234567890123456.12.azuredatabricks.net";
  const SERVER_ENDPOINT: &str = "1234567890123456.zerobus.eastus.azuredatabricks.net";
  const TABLE_NAME: &str = "main.default.air_quality";
  const CLIENT_ID: &str = "your-client-id";
  const CLIENT_SECRET: &str = "your-client-secret";


  #[tokio::main]
  async fn main() -> Result<(), Box<dyn Error>> {
      let sdk_handle = ZerobusSdk::builder()
          .endpoint(SERVER_ENDPOINT)
          .unity_catalog_url(DATABRICKS_WORKSPACE_URL)
          .build()?;

      let mut stream = sdk_handle
          .stream_builder()
          .table(TABLE_NAME)
          .oauth(CLIENT_ID, CLIENT_SECRET)
          .json()
          .max_inflight_requests(100)
          .build()
          .await?;

      stream.ingest_record_offset(
        JsonString("{
          \"device_name\": \"sensor\",
          \"temp\": 22,
          \"humidity\": 55}".to_string())).await?;

      println!("Record ingested successfully");
      stream.close().await?;
      println!("Stream closed successfully");

      Ok(())
  }

Protocol Buffers: For type-safe ingestion, use Protocol Buffers through .compiled_proto(descriptor) on the stream builder instead of .json(), where descriptor is a prost_types::DescriptorProto. Generate the needed files using the generate_proto tool and import into your project. 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. Enable with the Cargo feature: cargo add databricks-zerobus-ingest-sdk --features arrow-flight.

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

Java SDK

Java 8 or higher is required. The SDK provides low latency and efficient network I/O for high-throughput ingestion. It supports JSON (simplest) and Protocol Buffers (recommended for production).

Maven:

<dependency>
    <groupId>com.databricks</groupId>
    <artifactId>zerobus-ingest-sdk</artifactId>
    <version>0.2.0</version>
</dependency>

JSON example:

import com.databricks.zerobus.*;

public class ZerobusClient {

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
    private static final String SERVER_ENDPOINT =
        "https://1234567890123456.zerobus.eastus.azuredatabricks.net";
    private static final String DATABRICKS_WORKSPACE_URL =
        "https://adb-1234567890123456.12.azuredatabricks.net";
    private static final String TABLE_NAME = "main.default.air_quality";
    private static final String CLIENT_ID = "your-client-id";
    private static final String CLIENT_SECRET = "your-client-secret";

    public static void main(String[] args) throws Exception {
        ZerobusSdk sdk = new ZerobusSdk(
            SERVER_ENDPOINT,
            DATABRICKS_WORKSPACE_URL
        );

        ZerobusJsonStream stream = sdk.streamBuilder()
            .table(TABLE_NAME)
            .oauth(CLIENT_ID, CLIENT_SECRET)
            .json()
            .build()
            .join();

        try {
            for (int i = 0; i < 100; i++) {
                String record = String.format(
                    "{\"device_name\": \"sensor-%d\", \"temp\": 22, \"humidity\": 55}", i
                );
                stream.ingestRecordOffset(record);
            }
        } finally {
            stream.close();
        }
    }
}

Protocol Buffers: For type-safe ingestion, create a ZerobusProtoStream with streamBuilder() and .compiledProto(...). Generate a schema from your table using the bundled JAR tool, then compile it with protoc.

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.

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

Go SDK

Go 1.21 or higher is required. The SDK provides high throughput and performance for streaming ingestion. It supports JSON (simplest) and Protocol Buffers (recommended for production).

go get github.com/databricks/zerobus-sdk/go@latest

JSON example:

For simplicity, errors are ignored here. In production code, always check errors.

package main

import (
	"fmt"

	zerobus "github.com/databricks/zerobus-sdk/go"
)

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
const (
	ServerEndpoint         = "https://1234567890123456.zerobus.eastus.azuredatabricks.net"
	DatabricksWorkspaceURL = "https://adb-1234567890123456.12.azuredatabricks.net"
	TableName              = "main.default.air_quality"
	ClientID               = "your-client-id"
	ClientSecret           = "your-client-secret"
)

func main() {
	sdk, _ := zerobus.NewZerobusSdk(
		ServerEndpoint,
		DatabricksWorkspaceURL,
	)
	defer sdk.Free()

	options := zerobus.DefaultStreamConfigurationOptions()
	options.RecordType = zerobus.RecordTypeJson

	stream, _ := sdk.CreateStream(
		zerobus.TableProperties{
			TableName: TableName,
		},
		ClientID,
		ClientSecret,
		options,
	)
	defer stream.Close()

	_, _ = stream.IngestRecordOffset(`{
		"device_name": "sensor-001",
		"temp": 20,
		"humidity": 60
	}`)

  fmt.Println("Record ingested successfully")

  _ = stream.Close()
  fmt.Println("Stream closed successfully")
}

Protocol Buffers: For type-safe ingestion, use Protocol Buffers with RecordTypeProto (default) and provide a descriptorProto in table properties. Create a .proto file matching your table schema and run generate_proto script to help you import the files into your project.

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.

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

C++ SDK

Important

The C++ SDK is in Beta.

C++17 or higher is required. The SDK provides native gRPC streaming, OAuth, and automatic recovery through an RAII C++ interface. It supports JSON for simple setups and Protocol Buffers for production workloads.

The SDK ships as a prebuilt, per-platform release bundle, so you don't need a Rust toolchain to use it. Download the bundle for your platform (macOS, Linux including musl, or Windows) from the releases page, extract it, then point CMake at the bundled FFI archive. The archive is named libzerobus_ffi.a on macOS and Linux and zerobus_ffi.lib on Windows:

# macOS and Linux
cmake -S cpp -B build \
  -DZEROBUS_FFI_LIBRARY="$PWD/lib/libzerobus_ffi.a" \
  -DZEROBUS_FFI_HEADER_DIR="$PWD/lib"
cmake --build build -j

On Windows (PowerShell), point at the .lib archive instead:

cmake -S cpp -B build `
  -DZEROBUS_FFI_LIBRARY="$PWD/lib/zerobus_ffi.lib" `
  -DZEROBUS_FFI_HEADER_DIR="$PWD/lib"
cmake --build build -j

To build the SDK from a source checkout in your own CMake project, add it as a subdirectory and link the target. You can also use FetchContent to fetch it at configure time. This builds the FFI from Rust source, so it requires a Rust toolchain:

add_subdirectory(path/to/zerobus-sdk/cpp)
target_link_libraries(your_app PRIVATE zerobus::zerobus)

To consume a prebuilt bundle through add_subdirectory instead, set the FFI paths first so CMake links the bundled archive rather than trying to build it from Rust source that isn't present. Use zerobus_ffi.lib on Windows:

set(ZEROBUS_FFI_LIBRARY "/path/to/bundle/lib/libzerobus_ffi.a")
set(ZEROBUS_FFI_HEADER_DIR "/path/to/bundle/lib")
add_subdirectory(path/to/zerobus-sdk/cpp zerobus-cpp)
target_link_libraries(your_app PRIVATE zerobus::zerobus)

JSON example:

Ingestion is asynchronous and pipelined. The ingest_* methods queue a record and return immediately. Queue the batch and call flush() once, rather than waiting after each record.

#include "zerobus/zerobus.hpp"
#include <string>
#include <vector>

int main() {
  // See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
  const std::string SERVER_ENDPOINT = "https://1234567890123456.zerobus.eastus.azuredatabricks.net";
  const std::string DATABRICKS_WORKSPACE_URL = "https://adb-1234567890123456.12.azuredatabricks.net";
  const std::string TABLE_NAME = "main.default.air_quality";
  const std::string CLIENT_ID = "your-client-id";
  const std::string CLIENT_SECRET = "your-client-secret";

  zerobus::Sdk sdk = zerobus::Sdk::builder()
                         .endpoint(SERVER_ENDPOINT)
                         .unity_catalog_url(DATABRICKS_WORKSPACE_URL)
                         .application_name("my-app")
                         .build();

  zerobus::TableProperties table;
  table.table_name = TABLE_NAME;   // empty descriptor => JSON stream

  zerobus::StreamOptions options;
  options.record_type = zerobus::RecordType::Json;

  zerobus::Stream stream =
      sdk.create_stream(table, CLIENT_ID, CLIENT_SECRET, options);

  std::vector<std::string> batch = {
      R"({"device_name": "sensor-001", "temp": 20, "humidity": 60})",
      R"({"device_name": "sensor-002", "temp": 22, "humidity": 55})",
  };
  stream.ingest_json_records(batch);   // queue the batch — no per-record wait
  stream.flush();                      // wait once for all acks
  stream.close();

  return 0;
}

Every failure throws zerobus::ZerobusException, which carries a message and an is_retryable() flag. To track durability on a continuous stream without blocking, register an AckCallback via StreamOptions::ack_callback. Callbacks run serialized on a background thread and must be noexcept. See the C++ SDK docs for the full threading, drain-policy, and lifetime contract.

For type-safe ingestion, you can use Protocol Buffers in one of two ways:

Generate the schema from Unity Catalog with ProtoSchema::from_uc_json(). This builds a descriptor and JSON-to-proto encoder directly from the table's metadata, so it needs no .proto file or protoc:
1. Fetch the table's metadata JSON from the Get a table API (GET /api/2.1/unity-catalog/tables/{full_name}). The service principal needs SELECT on the table.
2. Pass the metadata to ProtoSchema::from_uc_json() to build the descriptor and encoder.
3. Set TableProperties::descriptor_proto, then ingest with ingest_proto_records().
Compile a checked-in .proto with protoc for compile-time typing.

For a runnable walkthrough, see the Protocol Buffers examples.

For columnar or batch-oriented ingestion of Apache Arrow record batches over the same gRPC connection, see Use Arrow Flight with Zerobus Ingest.

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

C# SDK

Important

The C# / .NET SDK is in Beta. The Databricks.Zerobus package is pre-release.

.NET 8.0 or higher is required. The SDK provides native gRPC streaming, OAuth, and automatic recovery. It supports JSON for simple setups and Protocol Buffers for production workloads. Arrow Flight is not available in the C# SDK.

Add the Databricks.Zerobus package to your project:

dotnet add package Databricks.Zerobus

JSON example:

using Databricks.Zerobus;

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
const string SERVER_ENDPOINT = "https://1234567890123456.zerobus.eastus.azuredatabricks.net";
const string DATABRICKS_WORKSPACE_URL = "https://adb-1234567890123456.12.azuredatabricks.net";
const string TABLE_NAME = "main.default.air_quality";
const string CLIENT_ID = "your-client-id";
const string CLIENT_SECRET = "your-client-secret";

using var sdk = ZerobusSdk.CreateBuilder()
    .Endpoint(SERVER_ENDPOINT)
    .UnityCatalogUrl(DATABRICKS_WORKSPACE_URL)
    .Build();

using var stream = sdk.CreateJsonStream(TABLE_NAME, CLIENT_ID, CLIENT_SECRET);

long offset = stream.IngestRecord(
    """{"device_name": "sensor-1", "temp": 22, "humidity": 55}""");
stream.WaitForOffset(offset);
stream.Close();

IngestRecord returns the record's offset, and WaitForOffset blocks until that record is durable. To ingest a batch, use IngestRecords, which takes an array of records and returns the last offset. Blocking on the offset is optional. See Message blocking and acknowledgment.

Protocol Buffers: For type-safe ingestion, create a stream with sdk.CreateProtoStream(TABLE_NAME, descriptorProto, CLIENT_ID, CLIENT_SECRET), where descriptorProto is the serialized DescriptorProto bytes for your compiled message, then ingest with stream.IngestRecord(protoBytes).

For complete documentation, configuration options, and Protocol Buffer examples, see the C# SDK repository.

TypeScript SDK

Node.js 16 or higher is required. The SDK provides high performance with async support through JavaScript Promises. It supports JSON (simplest) and Protocol Buffers (recommended for production).

npm install @databricks/zerobus-ingest-sdk

JSON example:

import { ZerobusSdk, RecordType } from '@databricks/zerobus-ingest-sdk';

// See "Get your workspace URL and Zerobus Ingest endpoint" for information on obtaining these values.
const SERVER_ENDPOINT = 'https://1234567890123456.zerobus.eastus.azuredatabricks.net';
const DATABRICKS_WORKSPACE_URL = 'https://adb-1234567890123456.12.azuredatabricks.net';
const TABLE_NAME = 'main.default.air_quality';
const CLIENT_ID = 'your-client-id';
const CLIENT_SECRET = 'your-client-secret';

const sdk = new ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL);

const stream = await sdk.createStream({ tableName: TABLE_NAME }, CLIENT_ID, CLIENT_SECRET, {
  recordType: RecordType.Json,
});

try {
  for (let i = 0; i < 100; i++) {
    const record = { device_name: `sensor-${i}`, temp: 22, humidity: 55 };
    await stream.ingestRecordOffset(record);
  }
} finally {
  await stream.close();
}

Protocol Buffers: For type-safe ingestion, use Protocol Buffers with RecordType.Proto (default) and provide a descriptorProto in table properties.

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.

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

REST API

The REST API allows you to ingest a single record by sending an HTTP POST request to the /zerobus/v1/tables/<table-name>/insert endpoint. The record itself is included in the request body and must be in JSON format.

This example walks you through how to use CURL to push data to Zerobus Ingest using the REST API.

Headers

The request requires two specific HTTP headers to authenticate and format the request correctly.

Content-Type: application/json
- Mandatory field for specifying the content type. Currently, JSON is the only supported message format.
Authorization: Bearer <token>
- Replace <token> with the OAuth token you have fetched using the curl command provided later.

Fetch OAuth Token: These tokens expire every hour and must be refreshed. You can refresh them by re-fetching the OAuth token.

Fill in the following parameters:

$CATALOG, $SCHEMA, $TABLE, $WORKSPACE_ID, $WORKSPACE_URL
$DATABRICKS_CLIENT_ID and $DATABRICKS_CLIENT_SECRET
- These two parameters correspond to the service principle you created.

authorization_details=$(cat <<EOF
[{
  "type": "unity_catalog_privileges",
  "privileges": ["USE CATALOG"],
  "object_type": "CATALOG",
  "object_full_path": "$CATALOG"
},
{
  "type": "unity_catalog_privileges",
  "privileges": ["USE SCHEMA"],
  "object_type": "SCHEMA",
  "object_full_path": "$CATALOG.$SCHEMA"
},
{
  "type": "unity_catalog_privileges",
  "privileges": ["SELECT", "MODIFY"],
  "object_type": "TABLE",
  "object_full_path": "$CATALOG.$SCHEMA.$TABLE"
}]
EOF
)

export OAUTH_TOKEN=$(curl -X POST \
  -u "$DATABRICKS_CLIENT_ID:$DATABRICKS_CLIENT_SECRET" \
  -d "grant_type=client_credentials" \
  -d "scope=all-apis" \
  -d "resource=api://databricks/workspaces/$WORKSPACE_ID/zerobusDirectWriteApi" \
  --data-urlencode "authorization_details=$authorization_details" \
  "$WORKSPACE_URL/oidc/v1/token" | jq -r '.access_token')

Record Ingestion:

Fill in the following parameters:

$ZEROBUS_ENDPOINT
- As defined in the Get your workspace URL and Zerobus Ingest Endpoint section.
$CATALOG, $SCHEMA, $TABLE, $WORKSPACE_ID, $WORKSPACE_URL
$OAUTH_TOKEN
- This was created in the previous step.

The request body must be a list of JSON objects.

curl -X POST \
  "$ZEROBUS_ENDPOINT/zerobus/v1/tables/$CATALOG.$SCHEMA.$TABLE/insert" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OAUTH_TOKEN" \
  -d '[{ "device_name": "device_num_1", "temp": 28, "humidity": 60 },
       { "device_name": "device_num_1", "temp": 28, "humidity": 60 }]'

If all of the information is filled in correctly, you should receive an empty JSON response with an HTTP status code of 200.

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:

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

Feedback

Was this page helpful?

Last updated on 2026-08-19

Use Zerobus Ingest

Get started with Zerobus Ingest

Choose an interface

Get your workspace URL and Zerobus Ingest endpoint

Create or identify the target table

Ingest into a streaming table

Create a service principal and grant permissions

Write a client

Python SDK

Rust SDK

Java SDK

Go SDK

C++ SDK

C# SDK

TypeScript SDK

REST API

Headers

Handle errors

Next steps

Feedback

Additional resources