databricks-zerobus-ingest-sdk 2.4.0

A high-performance Rust client for streaming data ingestion into Databricks Delta tables using the Zerobus service
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

//! # Databricks Zerobus Ingest SDK
//!
//! A high-performance Rust client for streaming data ingestion into Databricks Delta tables.
//!
//! ## Quick Start
//!
//! ```rust,ignore
//! use databricks_zerobus_ingest_sdk::{ZerobusSdk, JsonValue};
//!
//! let sdk = ZerobusSdk::builder()
//!     .endpoint(zerobus_endpoint)
//!     .unity_catalog_url(uc_endpoint)
//!     .build()?;
//!
//! let stream = sdk
//!     .stream_builder()
//!     .table("catalog.schema.table")
//!     .oauth(client_id, client_secret)
//!     .json()
//!     .build()
//!     .await?;
//!
//! // Idiomatic flow: ingest in a loop, then flush() once to confirm the batch.
//! for my_record in records {
//!     stream.ingest_record_offset(JsonValue(my_record)).await?;
//! }
//! stream.flush().await?; // Returns once every queued record is acknowledged.
//!
//! stream.close().await?;
//! ```
//!
//! See the `examples/` directory for complete working examples.

pub mod databricks {
    pub mod zerobus {
        include!(concat!(env!("OUT_DIR"), "/databricks.zerobus.rs"));
    }
}

#[cfg(feature = "arrow-flight")]
mod arrow_configuration;
#[cfg(feature = "arrow-flight")]
mod arrow_metadata;
#[cfg(feature = "arrow-flight")]
mod arrow_stream;
mod builder;
mod callbacks;
mod client_warnings;
mod default_token_factory;
mod errors;
mod headers_provider;
mod landing_zone;
#[cfg(feature = "testing")]
mod multiplexed_stream;
mod offset_generator;
mod proxy;
mod record_types;
pub mod schema;
mod sdk;
mod stream;
mod stream_configuration;
pub mod stream_options;
mod tls_config;
mod token_cache;

#[cfg(feature = "arrow-flight")]
pub use arrow_configuration::ArrowStreamConfigurationOptions;
#[cfg(feature = "arrow-flight")]
pub use arrow_stream::{ArrowSchema, DataType, Field, RecordBatch, TimeUnit, ZerobusArrowStream};
pub use builder::{StreamBuilder, ZerobusSdkBuilder};
pub use callbacks::AckCallback;
pub use default_token_factory::DefaultTokenFactory;
pub use errors::ZerobusError;
#[cfg(feature = "testing")]
pub use headers_provider::NoAuthHeadersProvider;
pub use headers_provider::{HeadersProvider, OAuthHeadersProvider};
#[cfg(feature = "testing")]
pub use multiplexed_stream::{MessageId, MultiplexedStream};
pub use offset_generator::{OffsetId, OffsetIdGenerator};
pub use proxy::{ConnectorFactory, ProxyConnector};
pub use record_types::{
    EncodedBatch, EncodedBatchIter, EncodedRecord, JsonEncodedRecord, JsonString, JsonValue,
    ProtoBytes, ProtoEncodedRecord, ProtoMessage,
};
pub use sdk::{ZerobusSdk, DEFAULT_SDK_IDENTIFIER};
#[cfg(feature = "testing")]
pub use stream::CallbackHandlerHarness;
pub use stream::ZerobusStream;
pub use stream_configuration::StreamConfigurationOptions;
#[cfg(feature = "testing")]
pub use tls_config::NoTlsConfig;
pub use tls_config::{SecureTlsConfig, TlsConfig};

#[cfg(feature = "zeroparser")]
pub mod zeroparser;

/// The type of the stream connection created with the server.
/// Currently we only support ephemeral streams on the server side, so we support only that in the SDK as well.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum StreamType {
    /// Ephemeral streams exist only for the duration of the connection.
    /// They are not persisted and are not recoverable.
    Ephemeral,
    /// UNSUPPORTED: Persistent streams are durable and recoverable.
    Persistent,
}

/// The properties of the table to ingest to.
///
/// Configure the table via the builder API:
/// `sdk.stream_builder().table("catalog.schema.table").compiled_proto(descriptor)`.
///
/// # Common errors:
/// -`InvalidTableName`: table_name contains invalid characters or doesn't exist
/// -`PermissionDenied`: insufficient permissions to write to the specified table
/// -`InvalidArgument`: invalid or missing descriptor_proto or auth token
#[derive(Debug, Clone)]
pub(crate) struct TableProperties {
    pub(crate) table_name: String,
    pub(crate) descriptor_proto: Option<prost_types::DescriptorProto>,
}

pub type ZerobusResult<T> = Result<T, ZerobusError>;