databricks-zerobus-ingest-sdk 2.2.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
125

mod common;

use common::{
    bench_prost_reflect_decode, bench_prost_typed_decode, bench_zeroparser_decode,
    create_encoded_messages, format_bytes, load_bench_sample, BenchmarkConfig,
};
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};

struct BenchScenario {
    group_name: &'static str,
    config: BenchmarkConfig,
    sample_json: String,
    padding_field: &'static str,
    message_sizes: &'static [usize],
    message_counts: &'static [usize],
}

fn air_quality_scenario() -> BenchScenario {
    BenchScenario {
        group_name: "air_quality_decode",
        config: BenchmarkConfig::for_message("AirQuality"),
        sample_json: r#"{"device_name":"sensor_01","temp":25,"humidity":65}"#.to_string(),
        padding_field: "device_name",
        message_sizes: &[32, 256, 1_024],
        message_counts: &[10_000, 100_000],
    }
}

fn clickbench_scenario() -> BenchScenario {
    BenchScenario {
        group_name: "clickbench_decode",
        config: BenchmarkConfig::for_message("ClickBench"),
        sample_json: load_bench_sample("click_bench_json"),
        padding_field: "URL",
        message_sizes: &[1_024],
        message_counts: &[100_000, 1_000_000],
    }
}

fn supported_nullable_types_scenario() -> BenchScenario {
    BenchScenario {
        group_name: "supported_nullable_types_decode",
        config: BenchmarkConfig::for_message("SupportedNullableTypes"),
        sample_json: load_bench_sample("supported_nullable_types_json"),
        padding_field: "description",
        message_sizes: &[128, 512, 1_024],
        message_counts: &[10_000, 100_000],
    }
}

fn run_decode_benchmark(c: &mut Criterion, scenario: BenchScenario) {
    let mut group = c.benchmark_group(scenario.group_name);

    for &target_size in scenario.message_sizes {
        for &count in scenario.message_counts {
            let encoded_messages = create_encoded_messages(
                &scenario.config.msg_desc,
                &scenario.sample_json,
                scenario.padding_field,
                target_size,
                count,
            );
            let actual_message_size = encoded_messages.first().map(|m| m.len()).unwrap_or(0);
            let total_bytes = actual_message_size * count;
            let description = format_args!(
                "{} size x {} messages",
                format_bytes(actual_message_size),
                count
            );
            group.throughput(Throughput::Bytes(total_bytes as u64));

            group.bench_with_input(
                BenchmarkId::new("prost_reflect", description),
                &encoded_messages,
                |b, messages| {
                    b.iter(|| bench_prost_reflect_decode(&scenario.config.msg_desc, messages));
                },
            );

            group.bench_with_input(
                BenchmarkId::new("prost", description),
                &encoded_messages,
                |b, messages| {
                    b.iter(|| bench_prost_typed_decode(scenario.config.prost_typed, messages));
                },
            );

            group.bench_with_input(
                BenchmarkId::new("zeroparser", description),
                &encoded_messages,
                |b, messages| {
                    b.iter(|| {
                        bench_zeroparser_decode(
                            &scenario.config.registry,
                            &scenario.config.fields,
                            messages,
                        )
                    });
                },
            );
        }
    }

    group.finish();
}

fn bench_air_quality(c: &mut Criterion) {
    run_decode_benchmark(c, air_quality_scenario());
}

fn bench_clickbench(c: &mut Criterion) {
    run_decode_benchmark(c, clickbench_scenario());
}

fn bench_supported_nullable_types(c: &mut Criterion) {
    run_decode_benchmark(c, supported_nullable_types_scenario());
}

criterion_group! {
    name = parser_benches;
    config = Criterion::default().sample_size(10);
    targets = bench_air_quality, bench_clickbench, bench_supported_nullable_types,
}

criterion_main!(parser_benches);