Loading...
Loading...
Build Zerobus Ingest clients for near real-time data ingestion into Databricks Delta tables via gRPC. Use when creating producers that write directly to Unity Catalog tables without a message bus, working with the Zerobus Ingest SDK in Python/Java/Go/TypeScript/Rust, generating Protobuf schemas from UC tables, or implementing stream-based ingestion with ACK handling and retry logic.
npx skill4agent add databricks/databricks-agent-skills databricks-zerobus-ingest| Scenario | Language | Serialization | Reference |
|---|---|---|---|
| Quick prototype / test harness | Python | JSON | references/2-python-client.md |
| Production Python producer | Python | Protobuf | references/2-python-client.md + references/4-protobuf-schema.md |
| JVM microservice | Java | Protobuf | references/3-multilanguage-clients.md |
| Go service | Go | JSON or Protobuf | references/3-multilanguage-clients.md |
| Node.js / TypeScript app | TypeScript | JSON | references/3-multilanguage-clients.md |
| High-performance system service | Rust | JSON or Protobuf | references/3-multilanguage-clients.md |
| Schema generation from UC table | Any | Protobuf | references/4-protobuf-schema.md |
| Retry / reconnection logic | Any | Any | references/5-operations-and-limits.md |
.protogrpcio-toolsprotobufprotobufgrpcio-tools==1.62.0MODIFYSELECTfrom zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import RecordType, StreamConfigurationOptions, TableProperties
sdk = ZerobusSdk(server_endpoint, workspace_url)
options = StreamConfigurationOptions(record_type=RecordType.JSON)
table_props = TableProperties(table_name)
stream = sdk.create_stream(client_id, client_secret, table_props, options)
try:
# Pass a dict for JSON streams; the SDK serializes it.
record = {"device_name": "sensor-1", "temp": 22, "humidity": 55}
offset = stream.ingest_record_offset(record)
stream.wait_for_offset(offset) # Block until durably written
finally:
stream.close()| Topic | File | When to Read |
|---|---|---|
| Setup & Auth | references/1-setup-and-authentication.md | Endpoint formats, service principals, SDK install |
| Python Client | references/2-python-client.md | Sync/async Python, JSON and Protobuf flows, reusable client class |
| Multi-Language | references/3-multilanguage-clients.md | Java, Go, TypeScript, Rust SDK examples |
| Protobuf Schema | references/4-protobuf-schema.md | Generate .proto from UC table, compile, type mappings |
| Operations & Limits | references/5-operations-and-limits.md | ACK handling, retries, reconnection, throughput limits, constraints |
.protoscripts/zerobus_ingest.pydatabricks workspace import-dir ./scripts /Workspace/Users/<user>/scriptsMODIFYSELECTauthorization_details<user>USER=$(databricks current-user me --output json | jq -r .userName)
databricks workspace import-dir ./scripts "/Workspace/Users/$USER/scripts"databricks jobs create --json '{
"name": "zerobus-ingest",
"tasks": [{
"task_key": "ingest",
"spark_python_task": {
"python_file": "/Workspace/Users/<user>/scripts/zerobus_ingest.py"
},
"new_cluster": {
"spark_version": "16.1.x-scala2.12",
"node_type_id": "i3.xlarge",
"num_workers": 0
}
}]
}'
databricks jobs run-now JOB_IDdatabricks workspace import-dir ./scripts /Workspace/Users/<user>/scriptsdatabricks jobs run-now JOB_IDdatabricks-zerobus-ingest-sdk"libraries": [
{"pypi": {"package": "databricks-zerobus-ingest-sdk>=1.0.0"}}
]TIMESTAMPint64from datetime import datetime, timezone
event_time = int(datetime.now(timezone.utc).timestamp() * 1_000_000) # epoch microsecondsingest_record_offset()wait_for_offset(offset)AckCallbackflush()| Issue | Solution |
|---|---|
| Connection refused | Verify server endpoint format matches your cloud (AWS vs Azure). Check firewall allowlists. |
| Authentication failed | Confirm service principal client_id/secret. Verify GRANT statements on the target table. |
| Schema mismatch | Ensure record fields match the target table schema exactly. Regenerate .proto if table changed. |
| Stream closed unexpectedly | Implement retry with exponential backoff and stream reinitialization. See references/5-operations-and-limits.md. |
| Throughput limits hit | Max 100 MB/s and 15,000 rows/s per stream. Open multiple streams or contact Databricks. |
| Region not supported | Check supported regions in references/5-operations-and-limits.md. |
| Table not found | Ensure table is a managed Delta table in a supported region with correct three-part name. |
| SDK install fails on serverless | The Zerobus SDK cannot be pip-installed on serverless compute. Use classic compute clusters or the REST API (Beta) from notebooks. |
| Error 4024 / authorization_details | Service principal lacks explicit table-level grants. Grant |