# 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`](pipeline-run-design.md); the sibling
> workflow docs are [`pipelines-guide.md`](pipelines-guide.md) and
> [`thund-guide.md`](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.
```sh
# 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`](pipeline-run-design.md).
### 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`:
```sh
# 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…`:
```rust
// 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?;
```
| param | effect |
|---|---|
| `user_id=alice` | `UserContext.user_id = "alice"` on every `ExecutePlan` |
| `session_id=sess-7` | the request `session_id` (omit → a generated `new_uuid`) |
| `token=…` | attached as `authorization: Bearer …` on every request |
| `use_ssl`, `user_agent`, … | recognised but **ignored** (not session-shaping yet) |
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.:
```sh
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).
```rust
// 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):
```sh
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.
```rust
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:
```sh
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`](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
| I want to… | command |
|---|---|
| watch the demo DAG in a browser (no server) | `cargo run -p knut-pipelines --features web --bin knutpipe -- web` |
| stream a live run into the browser | `… --features web,grpc … web --connect sc://…` |
| run the browser robot (needs a WebDriver) | `chromedriver --port=4444` then `cargo test -p knut-pipelines --features web --test robot_web_viewer -- --ignored --nocapture` |
| connect authenticated | `sc://host:15002/;user_id=U;token=T` on any grpc subcommand |
| emit Airflow asset events | build `--features airflow`, fold a run through `AirflowAssetSink`, `emit` |
| run a CDC/SCD merge | build `-p knut-thund --features native`, `FlowKind::Cdc { CdcSpec { … } }` |
| build the wasm guest for the gated test | `cargo run -p xtask -- build-wasm-guest` |