Use Arrow Flight with Zerobus Ingest

Important

Arrow Flight ingestion is in Beta.

Arrow Flight ingestion lets you send Apache Arrow RecordBatch data directly to Zerobus Ingest instead of converting every row to JSON or Protocol Buffers (protobuf) first. It is a third record format option on the Zerobus SDKs, alongside JSON and protobuf, and it runs over the same gRPC connection. It uses the same Zerobus endpoint, the same OAuth flow, and the same x-databricks-zerobus-table-name header convention. The wire protocol is Arrow Flight DoPut, which carries Arrow IPC messages over gRPC.

When to use Arrow Flight

Arrow Flight is the best fit in the following scenarios:

Your application already produces Arrow data, such as pyarrow.Table or pyarrow.RecordBatch (Python), arrow_array::RecordBatch from the arrow-rs crates (Rust), or VectorSchemaRoot (Java). DataFrame libraries built on Arrow, such as Polars or DataFusion, fit naturally into this path.
You ingest rows in batches instead of sending one record at a time.
Your schema is wide, numeric-heavy, or analytics-oriented, where row-by-row serialization adds noticeable CPU overhead.
You are building collectors or gateways that aggregate data for a short interval and then send it as one column format batch.

Arrow Flight is usually not the best choice for sparse, one-row-at-a-time traffic. In those cases, JSON or protobuf over the SDK gRPC path are typically simpler. See Choose an interface.

How the ingestion model works

With Arrow Flight ingestion, one stream writes to one target table. To ingest data, follow this sequence:

Define an Arrow schema that matches the destination Delta table schema.
Open a Zerobus Arrow stream for that table.
Send RecordBatch (or Table) payloads.
Wait for the last offset or call flush() to confirm durability.
Close the stream.

If you use a Zerobus SDK, the SDK handles the low-level Arrow Flight wire details for you. It serializes your Arrow data to IPC format and splits a large batch into smaller transport messages, which the server acknowledges individually.

Arrow Flight does not provide all-or-nothing durability for the whole logical batch. An Arrow batch can be very large, and because the SDK divides it into separate transport messages that are acknowledged as they are received, a large batch can be partially durable if a failure occurs partway through. This is different from JSON and protobuf batches, which commit atomically and are bounded by the 10 MB message size. See Arrow Flight batches are the exception.

The logical-offset abstraction still holds on top of this chunking. ingest_batch() returns a single logical offset for the batch you submitted, and wait_for_offset() on that offset completes only after every transport message that makes up the batch has been acknowledged. (Method names are from the Python SDK; other SDKs expose equivalent methods, such as ingestBatch and waitForOffset in Java.)

As with the Protobuf schema rule, the schema you pass to the stream must fit into the target Delta table: it must contain at minimum all of the non-nullable columns. Your schema may omit nullable columns that exist in the Delta table (this is treated as a non-breaking schema change), but any other mismatch is rejected. Each Arrow field's type must be compatible with its Delta column. For the supported Delta types, see Supported data types.

Because the SDK splits large batches into transport messages, an Arrow batch is not subject to the 10 MB message-size limit the way a JSON or Protocol Buffers batch is. Because Arrow Flight runs over the same gRPC transport, the same throughput, latency, and quota characteristics apply, all of which scale to meet higher workloads. See Zerobus Ingest quotas.

Write a client

The examples below open an Arrow Flight stream against the same air_quality table used in the Use Zerobus Ingest examples. They are shown in Python and Rust for brevity, but the same builder, configuration options, and call sequence are available in every Zerobus SDK. Adapt the syntax for your language and consult the SDK repository for language-specific Arrow types.

Python SDK

The Python SDK accepts a pyarrow.Schema at stream creation time and a pyarrow.RecordBatch or pyarrow.Table for each ingest call.

pip install "databricks-zerobus-ingest-sdk[arrow]" pyarrow

import pyarrow as pa

from zerobus.sdk.sync import ZerobusSdk

# See "Get your workspace URL and Zerobus Ingest endpoint" in zerobus-ingest.md.
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"

schema = pa.schema(
    [
        ("device_name", pa.large_utf8()),
        ("temp", pa.int32()),
        ("humidity", pa.int64()),
    ]
)

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

stream = sdk.create_arrow_stream(TABLE_NAME, schema, CLIENT_ID, CLIENT_SECRET)

row_count = 1_000
batch = pa.record_batch(
    {
        "device_name": [f"sensor-{i}" for i in range(row_count)],
        "temp": [20 + (i % 5) for i in range(row_count)],
        "humidity": [55 + (i % 10) for i in range(row_count)],
    },
    schema=schema,
)

try:
    offset = stream.ingest_batch(batch)

    # Optional: block until the batch is durably written
    stream.wait_for_offset(offset)
finally:
    stream.close()

stream.ingest_batch() also accepts a pyarrow.Table. The SDK converts it to a single RecordBatch internally before sending. Each call returns a logical offset. Blocking on the offset is optional. For when to wait and how acknowledgment works, see Message blocking and acknowledgment.

Rust SDK

The Rust SDK exposes Arrow Flight through the stream_builder() API, behind the arrow-flight Cargo feature. Use the same Arrow major version as the SDK so the RecordBatch and array types match at compile time.

cargo add databricks-zerobus-ingest-sdk --features arrow-flight
cargo add arrow-array
cargo add arrow-schema
cargo add tokio --features macros,rt-multi-thread

use std::sync::Arc;

use arrow_array::{Int32Array, Int64Array, LargeStringArray, RecordBatch};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use databricks_zerobus_ingest_sdk::ZerobusSdk;

const SERVER_ENDPOINT: &str = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com";
const DATABRICKS_WORKSPACE_URL: &str = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com";
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 std::error::Error>> {
    let schema = Arc::new(ArrowSchema::new(vec![
        Field::new("device_name", DataType::LargeUtf8, false),
        Field::new("temp", DataType::Int32, false),
        Field::new("humidity", DataType::Int64, false),
    ]));

    let sdk = ZerobusSdk::builder()
        .endpoint(SERVER_ENDPOINT)
        .unity_catalog_url(DATABRICKS_WORKSPACE_URL)
        .build()?;

    let mut stream = sdk
        .stream_builder()
        .table(TABLE_NAME)
        .oauth(CLIENT_ID, CLIENT_SECRET)
        .arrow(Arc::clone(&schema))
        .build_arrow()
        .await?;

    let row_count: i32 = 1_000;
    let batch = RecordBatch::try_new(
        Arc::clone(&schema),
        vec![
            Arc::new(LargeStringArray::from(
                (0..row_count)
                    .map(|i| format!("sensor-{i}"))
                    .collect::<Vec<_>>(),
            )),
            Arc::new(Int32Array::from(
                (0..row_count).map(|i| 20 + (i % 5)).collect::<Vec<_>>(),
            )),
            Arc::new(Int64Array::from(
                (0..row_count)
                    .map(|i| 55 + (i % 10) as i64)
                    .collect::<Vec<_>>(),
            )),
        ],
    )?;

    let offset = stream.ingest_batch(batch).await?;

    // Optional: block until the batch is durably written
    stream.wait_for_offset(offset).await?;
    stream.close().await?;

    Ok(())
}

The builder selects the Arrow Flight format with .arrow(schema) and finalizes the stream with .build_arrow(), which returns a ZerobusArrowStream. JSON and protobuf continue to use .json() / .compiled_proto(...) and .build().

Ingesting VARIANT columns

Apache Arrow has no native VARIANT type. To ingest into a VARIANT column over Arrow Flight, build the column's backing metadata and value fields as a struct of two LargeBinary columns, then include that struct in your RecordBatch. Over the gRPC SDKs and REST, you instead pass a Variant value as a JSON-encoded string. See Supported data types.

The following Rust example builds a VARIANT struct column from JSON rows and ingests it:

fn variant_struct(json_rows: &[&str]) -> ArrayRef {
    let mut metas: Vec<Vec<u8>> = Vec::new();
    let mut vals: Vec<Vec<u8>> = Vec::new();
    for json in json_rows {
        let mut vb = VariantBuilder::new();
        vb.append_json(json).expect("invalid JSON for variant");
        let (metadata, value) = vb.finish();
        metas.push(metadata);
        vals.push(value);
    }
    let fields = Fields::from(vec![
        Field::new("metadata", DataType::LargeBinary, false),
        Field::new("value", DataType::LargeBinary, false),
    ]);
    let meta_arr = Arc::new(LargeBinaryArray::from_iter_values(metas)) as ArrayRef;
    let val_arr = Arc::new(LargeBinaryArray::from_iter_values(vals)) as ArrayRef;
    Arc::new(StructArray::try_new(fields, vec![meta_arr, val_arr], None).expect("variant struct"))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client_id = std::env::var("DATABRICKS_CLIENT_ID")?;
    let client_secret = std::env::var("DATABRICKS_CLIENT_SECRET")?;

    let variant_type = DataType::Struct(Fields::from(vec![
        Field::new("metadata", DataType::LargeBinary, false),
        Field::new("value", DataType::LargeBinary, false),
    ]));
    let schema = Arc::new(ArrowSchema::new(vec![
        Field::new("id", DataType::Int32, true),
        Field::new("payload", variant_type, true),
    ]));

    let sdk = ZerobusSdk::builder()
        .endpoint(ENDPOINT)
        .unity_catalog_url(UC_URL)
        .build()?;

    let mut stream = sdk
        .stream_builder()
        .table(TABLE)
        .oauth(&client_id, &client_secret)
        .arrow(schema.clone())
        .ipc_compression(None)
        .build_arrow()
        .await?;

    let ids = Int32Array::from(vec![1, 2, 3]);
    let payload = variant_struct(&[
        r#"{"user":"alice","tags":[1,2,3]}"#,
        r#""just a string""#,
        r#"{"nested":{"a":true,"b":null,"c":3.14}}"#,
    ]);
    let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(ids) as ArrayRef, payload])?;

    let offset = stream.ingest_batch(batch).await?;
    stream.flush().await?;
    stream.close().await?;
    Ok(())
}

This example is in Rust. For equivalent usage in other languages, see the Zerobus SDK repository.

IPC compression

By default, Arrow IPC payloads are sent uncompressed. You can optionally compress them on the wire using one of two codecs.

LZ4_FRAME: Fast, low CPU overhead, modest compression ratio. Prefer this when the client is CPU-constrained but still wants to reduce bytes on the wire.
ZSTD: Higher compression ratio, more CPU per batch. Enable it whenever your client can absorb the additional CPU cost.

Compression reduces bytes on the wire but adds CPU cost on the client. Smaller payloads can avoid network bottlenecks and reduce network costs.

In the Python SDK, set the ipc_compression field on ArrowStreamConfigurationOptions:

from zerobus.sdk.shared.arrow import IPCCompression, ArrowStreamConfigurationOptions

options = ArrowStreamConfigurationOptions(ipc_compression=IPCCompression.ZSTD)

In the Rust SDK, set it on the builder. The CompressionType enum lives in the arrow-ipc crate, so add it as a dependency:

cargo add arrow-ipc

use arrow_ipc::CompressionType;

let stream = sdk
    .stream_builder()
    .table(TABLE_NAME)
    .oauth(CLIENT_ID, CLIENT_SECRET)
    .arrow(schema)
    .ipc_compression(Some(CompressionType::ZSTD))
    .build_arrow()
    .await?;

Best practices

Follow these guidelines to get the best performance and reliability from Arrow Flight ingestion.

Reuse a stream for many batches instead of opening a new stream per batch. Stream creation carries a significant overhead that you can amortize by reusing a stream across many batches.
Send multiple rows per batch. Start with natural application-sized batches, not one row per call. Sending one row at a time works, but negates most of the performance advantage of using Arrow.
Call flush() at controlled checkpoints. This gives you a clear durability boundary for a group of batches without blocking on every single one.
Enable IPC compression to improve throughput. ZSTD is recommended for most workloads when the client has spare CPU. Use LZ4_FRAME or no compression if the client is CPU-constrained.
Use Arrow Flight when your producer is already columnar. If your source data is naturally row-oriented and small, using Zerobus Ingest with JSON or protobuf is often simpler. See Use Zerobus Ingest.

Error handling and recovery

Arrow Flight streams use the same gRPC error categories as the rest of Zerobus Ingest. For error codes, retry guidance, and the full client-vs-server taxonomy, see Zerobus Ingest error handling.

When you configure the SDK with automatic recovery (the default), it transparently reconnects and replays unacknowledged batches on transient failures. After the stream closes, you can retrieve any batches that the server received but did not yet acknowledge. This applies whether the stream closed gracefully or due to an unrecoverable failure. In the Python SDK:

# Retry unacked_batches against a freshly created stream
if stream.is_closed:
    unacked_batches = stream.get_unacked_batches()

In the Rust SDK, call stream.get_unacked_batches().await? to retrieve unacknowledged batches for retry.

Additional resources

Use Zerobus Ingest: If you haven't set up Zerobus Ingest yet, start here for instructions on finding your workspace URL, creating the target Delta table, and configuring a service principal. These steps are shared across all record formats.
Zerobus Ingest quotas: Review the default Zerobus quotas before deploying to production. The same throughput and latency characteristics apply to Arrow Flight, and scale to meet higher workloads.
Zerobus Ingest error handling: Consult this page for a full list of gRPC error codes and recommended retry and recovery behavior for your client

Feedback

Was this page helpful?

Last updated on 2026-08-19