Use Arrow Flight with Zerobus Ingest

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 in Zerobus SDKs that support Arrow Flight, 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:

  1. Define an Arrow schema that matches the destination Delta table schema.
  2. Open a Zerobus Arrow stream for that table.
  3. Send RecordBatch (or Table) payloads.
  4. Wait for the last offset or call flush() to confirm durability.
  5. 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 automatically splits a large batch into smaller, ordered Flight messages.

The server reports cumulative progress as records become durable. A large logical batch can therefore be partially durable if a failure occurs while it is being sent. ingest_batch() still returns one logical offset for the submitted batch, and waiting for that offset confirms that all of its records are durable. See Arrow Flight batches are the exception.

Zerobus matches Arrow fields to Delta columns by name. The Arrow schema must include all required (non-nullable) Delta columns. You can omit nullable columns, which Zerobus writes as NULL. Do not include fields that are absent from the target table. Included fields must follow the Delta schema's relative order, match its nullability, and use the target column's type. For details, see schema matching rules.

An Arrow batch is not subject to the 10 MB gRPC message size limit. However, each individual row inside a RecordBatch must fit within the 10 MB limit. The SDK automatically splits larger batches into multiple wire messages, but it cannot split one oversized row. See Zerobus Ingest quotas.

Write a client

The examples below use the Python and Rust SDKs. For other languages that support Arrow Flight, see the Zerobus SDK repository.

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]"

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)

try:
    for start in range(0, 10_000, 1_000):
        end = start + 1_000
        batch = pa.record_batch(
            {
                "device_name": [f"sensor-{i}" for i in range(start, end)],
                "temp": [20 + (i % 5) for i in range(start, end)],
                "humidity": [55 + (i % 10) for i in range(start, end)],
            },
            schema=schema,
        )
        stream.ingest_batch(batch)
    stream.flush()
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. The example creates several batches and calls flush() once to confirm that all pending batches are durable. Use wait_for_offset() when you need to confirm a specific batch before continuing; waiting for the last offset also confirms all earlier offsets. 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, true),
        Field::new("temp", DataType::Int32, true),
        Field::new("humidity", DataType::Int64, true),
    ]));

    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?;

    for start in (0_i32..10_000).step_by(1_000) {
        let end = start + 1_000;
        let batch = RecordBatch::try_new(
            Arc::clone(&schema),
            vec![
                Arc::new(LargeStringArray::from(
                    (start..end)
                        .map(|i| format!("sensor-{i}"))
                        .collect::<Vec<_>>(),
                )),
                Arc::new(Int32Array::from(
                    (start..end).map(|i| 20 + (i % 5)).collect::<Vec<_>>(),
                )),
                Arc::new(Int64Array::from(
                    (start..end)
                        .map(|i| 55 + (i % 10) as i64)
                        .collect::<Vec<_>>(),
                )),
            ],
        )?;
        stream.ingest_batch(batch).await?;
    }
    stream.flush().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.

Python SDK

Set the ipc_compression field on ArrowStreamConfigurationOptions:

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

options = ArrowStreamConfigurationOptions(ipc_compression=IPCCompression.ZSTD)
stream = sdk.create_arrow_stream(
    TABLE_NAME, schema, CLIENT_ID, CLIENT_SECRET, options=options
)

Rust SDK

Set the compression type on the stream 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 a stream closes with unacknowledged work, the SDK retains batches that the client accepted but the server did not acknowledge, including batches that might not have been sent yet.

After automatic recovery is exhausted, fix the cause of the failure and call close() to finalize the stream's unacknowledged batches. Because the stream has already failed, close() might return the same terminal error even though finalization succeeds.

You can call get_unacked_batches() only after the stream is closed. It returns the retained batches for persistence or application-managed replay. How you create a replacement stream, persist the batches, and retry them depends on your application's recovery policy.

Python SDK

from zerobus.sdk.shared import ZerobusException

try:
    stream.close()
except ZerobusException:
    # The terminal error can be returned after closure is finalized.
    pass

unacked_batches = stream.get_unacked_batches()

Rust SDK

// The terminal error can be returned after closure is finalized.
let _ = stream.close().await;
let unacked_batches = stream.get_unacked_batches().await?;

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 Zerobus quotas before deploying to production. Throughput, latency, and partitioned-table limits all apply to Arrow Flight.
  • Zerobus Ingest error handling: Consult this page for a full list of gRPC error codes and recommended retry and recovery behavior for your client.
  • Last updated on