nordisk/knut

Running a pipeline end-to-end — guide

Class: guide · audience: a techie who wants to run and watch a Spark Declarative Pipeline, connect an authenticated sc:// session, emit Airflow asset events, and run a native CDC/SCD merge. The what is in pipeline-run-design.md; the sibling workflow docs are pipelines-guide.md and thund-guide.md.

1 · The web viewer

The web feature serves facett's Rust facett-pipeline DAG viewer, compiled to wasm (the facet-wrapped bundle), and streams it PipelineModel frames over Server-Sent Events. knut draws nothing in JavaScript — the only script on the page is facet-wrapped's own loader, served as bytes. It demos with no Spark server — it replays a synthetic run so the DAG animates.

The bundle comes from a directory you name (--facet-dist DIR or FACET_WRAPPED_DIST=DIR) holding index.html, facet_wrapped.js and facet_wrapped_bg.wasm. In the facett checkout: cargo xtask wrap-wasm writes the two wasm files to facet-wrapped/dist/; put pkg-facet-wrapped/src/facet_wrapped_web/assets/index.html beside them. Without a bundle the server still starts, but GET / answers 503 with that recipe instead of a page — never a blank one.

# demo: serve the built-in bronze→silver→gold graph, synthetic replay
FACET_WRAPPED_DIST=../facett/facet-wrapped/dist \
cargo run -p knut-pipelines --features web --bin knutpipe -- web
# → # knutpipe web viewer on http://127.0.0.1:8787
#   open that URL: the Rust viewer's nodes go idle → running → done as the run replays.

# serve your own spec (edges are completed from each flow's query):
cargo run -p knut-pipelines --features web --bin knutpipe -- \
    web crates/knut-pipelines/examples/pipeline.json --addr 127.0.0.1:9000

# live: stream a REAL Spark Connect run into the page (needs web + grpc):
cargo run -p knut-pipelines --features web,grpc --bin knutpipe -- \
    web --connect sc://localhost:15002

The endpoints, if you want to drive it from a script instead of a browser: GET / (redirects to the loader: /index.html?component=pipeline&events=/events), the bundle files by name, GET /graph.json (the DAG, same bytes as dump_json), GET /pipeline.json (one all-idle PipelineModel frame on the wire) and GET /events (an SSE stream of frames — to_pipeline_model encoded by facett_pipeline::wire, {"title","stages":[{"id","name","status","kind","error", "metrics":{"rows","throughput_rows_s","lag_s"}}],"edges":[{"from","to","label"}]} — terminated by a {"done":true} frame). See pipeline-run-design.md#f1.

The browser robot test

The page has a real browser test (feature web, #[ignore]): it serves the real server on an ephemeral port and drives Chrome/Firefox through a WebDriver, gating on "every node reaches st-done". It is #[ignore] because it needs a WebDriver listening at 127.0.0.1:4444:

# 1. start a WebDriver (either one):
chromedriver --port=4444        # or:  geckodriver --port 4444

# 2. run the ignored browser drive:
cargo test -p knut-pipelines --features web --test robot_web_viewer \
    -- --ignored --nocapture
# → prints a ✓/✗ matrix and "── N/N robot steps PASSED ──"

The seam that makes this raceless is web::serve_on (web.rs:152): it takes an already-bound TcpListener, so the test binds 127.0.0.1:0, reads back the OS-assigned port via local_addr(), and hands the live listener to the server — fantoccini then navigates to a URL that is guaranteed to be listening.

2 · Connecting over sc:// with a user and token

PipelinesClient::connect (client.rs:206) parses the Spark Connect URL's session params and threads them through every request. The grammar is sc://host:port/;k=v;k=v…:

// user_id → UserContext.user_id on every request;
// session_id → pins the session (else a fresh UUID);
// token → `authorization: Bearer <token>` metadata via an interceptor.
let mut client = PipelinesClient::connect(
    "sc://spark.internal:15002/;user_id=alice;session_id=sess-7;token=eyJhbGc…",
).await?;

A plain http(s)://localhost:15002 connects with a generated session and no auth — the interceptor is a pass-through when there is no token. On the CLI, any subcommand that takes an sc:// URL (watch, pipeline, web --connect) accepts these params unchanged, e.g.:

cargo run -p knut-pipelines --features grpc --bin knutpipe -- \
    watch "sc://spark:15002/;user_id=alice;token=$SPARK_TOKEN" \
    --sql "SELECT count(*) FROM range(0, 1e8)"

Progress is server-driven. You do not ask Spark for progress — it pushes execution_progress (base.proto field 18) on its own while a plan runs, paced by the server's spark.connect.progress.reportInterval. execute_sql just consumes those frames, which is what makes watch --sql … a live Spark job monitor. (The live test knutpipe_live_path_streams_spark_job_progress asserts they arrive.)

3 · Emitting Airflow (AIP-82) asset events

Build with --features airflow. AirflowAssetSink is a PipelineEventSink: feed it a run's events and it collects one Airflow asset event per completed dataset; emit forwards the batch through any AssetEmitter (e.g. bifröst's HTTP emitter).

// cargo add / build: -p knut-pipelines --features airflow  (+ grpc to run live)
use knut_pipelines::AirflowAssetSink;

// URIs form as `{prefix}{dataset}`, e.g. sdp://sales/gold_daily
let mut sink = AirflowAssetSink::new(graph.clone(), "sdp://sales/");

// It IS the sink start_run pushes into — fold a run through it:
client.start_run(&run, &mut sink).await?;      // client.rs:373

// Collection is sync; emission is a separate async step:
for ev in sink.events() {                      // one per completed dataset
    println!("{} · {} rows", ev.uri, ev.extra["rows"]);
}
sink.emit(&airflow_emitter).await?;            // POST the batch (bifröst's transport)

Rules the tests pin: each output emits at most once; a failed output emits nothing (its asset is not current); emit is a no-op when nothing completed. The extra payload carries dataset / rows / output_type / source: "spark-sdp", and the wire body is bifröst's asset_event_body. It reuses bifröst's AssetEvent / AssetEmitter — no duplicate model.

Run its tests (default feature set — the mock emitter needs no server):

cargo test -p knut-pipelines --features airflow airflow::

4 · Running a native CDC / SCD pipeline

The native backend (knut-thund, feature native) now applies a change-data-capture merge over DataFusion. Your changelog must carry a _change_type column (INSERT / UPDATE_AFTER upsert · DELETE remove · UPDATE_BEFORE dropped) plus the key and sequence_by columns — the shared vocabulary skade and bifröst produce.

use knut_thund::ir::{CdcSpec, Flow, FlowKind, ScdType};

let flow = Flow {
    name: "apply_customers".into(),
    target: "customers".into(),
    reads: vec!["customer_changes".into()],       // the changelog source
    kind: FlowKind::Cdc { cdc: CdcSpec {
        keys: vec!["id".into()],                    // match key(s)
        sequence_by: "seq".into(),                  // change order
        apply_as_deletes: None,                     // or Some("op = 'D'")
        scd_type: ScdType::Type2,                   // Type1 = overwrite; Type2 = history
    }},
    query: Some(
        "SELECT id, name, seq, _change_type FROM customer_changes".into()),
    expectations: vec![],
};

• SCD type 1 — the target is the latest non-deleted row per key (history overwritten). _change_type is stripped from the output.
• SCD type 2 — every version is kept with derived __start_at / __end_at bounds (__end_at = NULL for the current version); a delete closes the open version.

If the changelog source was seeded as a micro-batch stream, the merge re-runs per trigger over the accumulated changelog (deterministic, since apply-changes is a pure function of the ordered changelog); otherwise it runs once over the bounded changelog. Run the operator's tests:

cargo test -p knut-thund --features native \
    cdc_scd_type1_upserts_and_deletes cdc_scd_type2_keeps_version_history

SparkBackend runs CDC only on spark_4_2(); the native backend runs it everywhere. See the capability matrix in thund-guide.md.

5 · Where the changelog comes from (bifröst → skade)

You don't build the _change_type changelog by hand for an Iceberg source: it is now single-sourced from skade. knut-bifrost's IcebergSource (feature skade) deleted its private DeltaPlan and delegates to skade::Table::read_changelog (source.rs:473), re-exporting the canonical skade::DeltaPlan (source.rs:355). skade walks the manifest lineage, tags appended rows INSERT and equality-deleted identities DELETE, and re-reads the whole snapshot on an overwrite window. bifröst strips the tag on a pure-append/full-reload window (rows go straight to the upsert pipeline) and keeps INSERT/DELETE tags only when the window actually carries deletes. Net effect: the vocabulary the thund CDC merge (§4) consumes is defined once, in skade, and never diverges.

Command cheat-sheet