spate 0.1.0

High-performance at-least-once ETL pipeline framework with a chaining operator API, checkpoint-driven source commits, sharded asynchronous sinks, backpressure, and Prometheus metrics.
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

//! NDJSON pipeline entirely in memory — no external systems.
//!
//! Shows the JSON deserializer decoding **N records from one payload** (JSON
//! Lines), with a malformed line skipped-and-counted
//! (`spate_json_deser_records_dropped_total`) and a filtered record dropped,
//! while the source offset only commits once every surviving record is durably
//! written (at-least-once). Runs anywhere via `spate-test`'s in-memory pieces:
//!
//! ```sh
//! cargo run -p spate --example json_ndjson_memory
//! ```

// Examples talk to their user on stdout/stderr by design.
#![allow(clippy::print_stdout, clippy::print_stderr)]

use serde::Deserialize;
use spate::json::{JsonDeserializerBuilder, JsonFraming, JsonSettings, OnError};
use spate::prelude::*;
use spate::source::LaneId;
use spate_test::{TestEncoder, capture_sink, memory_source};
use std::time::{Duration, Instant};

const CONFIG: &str = r#"
pipeline: { name: json-ndjson-demo, threads: 1 }
checkpoint: { interval: 200ms }
source: { memory: {} }
sink: { capture: {} }
"#;

#[derive(Debug, Deserialize)]
struct Event {
    #[expect(dead_code, reason = "decoded but not projected into the sink row")]
    id: u64,
    name: String,
    value: i64,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    spate::telemetry::init(spate::telemetry::LogFormat::Pretty, "info");
    let pipeline = Pipeline::from_config(PipelineConfig::from_str(CONFIG)?)?;

    let (source, handle) = memory_source();

    let (sink, script) = capture_sink(1, 1);
    let pool_cfg = {
        let mut cfg = SinkPoolConfig::default();
        cfg.batch.linger = Duration::from_millis(50); // flush quickly for the demo
        cfg
    };
    let sink = sink.with_pool_config(pool_cfg);

    // NDJSON in → typed `Event` → filter → project to a row → capturing sink.
    let runtime = pipeline
        .sink(sink)?
        .chains(|ctx| {
            let chunk_cfg = ctx.chunk();
            let deser = JsonDeserializerBuilder::from_settings(JsonSettings {
                framing: JsonFraming::Ndjson,
                on_error: OnError::Skip,
                reject_duplicate_keys: false,
            })
            .with_metrics(ctx.pipeline.clone(), "main")
            .build_serde::<Event>();
            chain_owned::<Event, _>(deser)
                .with_metrics(ctx.pipeline, "main")
                .filter(|e: &Event| e.value >= 0)
                .map(|e: Event| format!("{}={}", e.name, e.value).into_bytes())
                .sink(
                    TestEncoder,
                    KeyHashRouter,
                    chunk_cfg,
                    ctx.queues,
                    ctx.budget,
                )
                .build()
        })
        .runtime_options(RuntimeOptions {
            handle_signals: false,
            ..RuntimeOptions::default()
        })
        .into_runtime(source)?;
    let shutdown = runtime.shutdown_handle();
    let join = std::thread::spawn(move || runtime.run());

    // One payload, four NDJSON lines: line 2 is malformed (skipped), line 3
    // has a negative value (filtered out), lines 1 and 4 survive.
    let p0 = PartitionId(0);
    handle.assign_lanes(&[(LaneId(0), p0)]);
    let payload = concat!(
        "{\"id\":1,\"name\":\"alpha\",\"value\":10}\n",
        "NOT JSON\n",
        "{\"id\":2,\"name\":\"beta\",\"value\":-5}\n",
        "{\"id\":3,\"name\":\"gamma\",\"value\":30}"
    )
    .as_bytes();
    let last = handle.push(p0, Some(b"demo"), payload);

    let deadline = Instant::now() + Duration::from_secs(10);
    while handle.last_committed(p0) != Some(last + 1) {
        assert!(Instant::now() < deadline, "commit not observed in time");
        std::thread::sleep(Duration::from_millis(20));
    }
    shutdown.trigger();
    let report = join.join().expect("pipeline thread")?;

    let rows: Vec<String> = script
        .writes()
        .iter()
        .flat_map(|w| spate_test::decode_rows(&w.payload))
        .map(|r| String::from_utf8_lossy(&r).into_owned())
        .collect();

    println!("\npipeline exit: {:?}", report.state);
    println!("final watermarks: {:?}", report.final_watermarks);
    println!("rows written ({}): {rows:?}", rows.len());
    assert_eq!(
        rows.len(),
        2,
        "4 NDJSON lines: 1 malformed skipped, 1 negative filtered"
    );
    assert!(rows.contains(&"alpha=10".to_string()));
    assert!(rows.contains(&"gamma=30".to_string()));
    Ok(())
}