# knut-pipelines — guide
> **Class:** `guide` · audience: a techie driving `knutpipe` or the `PipelinesClient` API.
> The *what* is in [`pipelines-design.md`](pipelines-design.md); the *why* in [`pipelines-reasoning.md`](pipelines-reasoning.md).
> This file is the workflow.
## The `knutpipe` CLI
Ships `src/bin/knutpipe.rs`. Four subcommands; `--json-logs` is global.
| command | does | feature |
|---|---|---|
| `knutpipe show [graph.json] [--json]` | render a static graph (or a built-in demo); `--json` emits the spec format | default |
| `knutpipe watch <sc-url> [--graph ID \| --sql "…"] [--full-refresh]` | connect + stream: an existing run by id, or a plain SQL job's progress | `grpc` |
| `knutpipe gui [graph.json] [--connect sc://…]` | desktop egui + `egui-snarl` node graph | `gui` |
| `knutpipe pipeline <sc-url> --spec graph.json --storage file://… [--gui]` | define **and** run an SDP, streaming its lifecycle to the terminal (or `--gui`) | `grpc` (+`gui` for `--gui`) |
```sh
# static, no server:
knutpipe show # built-in bronze→silver→gold demo
knutpipe show pipeline.json --json
# live (build with --features grpc):
knutpipe watch sc://localhost:15002 --graph <GRAPH_ID>
knutpipe watch sc://localhost:15002 --sql "SELECT count(*) FROM range(0, 1e8)"
# define + run + watch (grpc):
knutpipe pipeline sc://localhost:15002 \
--spec crates/knut-pipelines/examples/pipeline.json \
--storage file:///tmp/knut_pipeline
# same, as a live node graph (needs --features gui):
knutpipe pipeline sc://localhost:15002 --spec …/pipeline.json \
--storage file:///tmp/knut_pipeline --gui
```
Two flags that bite the first time: a flow only runs if it carries a **`query`**
(else it is a drawn edge), and `--storage` is **required** to materialise outputs
(Spark needs a checkpoint/metadata location; must be writable by the workers).
## The Rust API
```rust
use knut_pipelines::{
DataflowGraph, Dataset, Flow, OutputType, PipelineRun,
PipelinesClient, ViewerState, ViewerUpdate, viewer,
};
// 1. connect (feature grpc) — sc:// or http(s)://
let mut client = PipelinesClient::connect("sc://localhost:15002").await?;
// 2a. watch a plain Spark SQL job — progress only, no pipeline:
let mut stream = client.execute_sql("SELECT count(*) FROM range(0, 1e8)").await?;
// 2b. or run an existing graph and merge events + progress into one stream:
let run = PipelineRun { graph_id: Some(id), full_refresh_all: true,
storage: Some("file:///tmp/pipeline".into()), ..Default::default() };
let mut stream = client.run_updates(&run).await?;
// 3. fold the stream into the single source of truth:
let mut state = ViewerState::new(graph);
use futures::StreamExt;
while let Some(update) = stream.next().await {
state.apply(update?); // returns true on real change
print!("{}", viewer::dump_state(&state, 5)); // render — CLI parity
}
```
Key items ([`pipelines-design.md`](pipelines-design.md) has the full pointers):
`PipelinesClient::connect` (`client.rs:205`), `run_updates` (`:449`),
`execute_sql` (`:475`), `define_graph` (`:348`); `ViewerState::apply`
(`progress.rs:240`), `state_json` (`:335`); `viewer::dump_state`/`dump_text`/`dump_json`
(`viewer.rs:88`,`:28`,`:18`).
## Worked example — define → run
A flow becomes runnable by attaching the SQL that defines it. `define_graph`
registers it (`CreateDataflowGraph` + `DefineOutput` + `DefineFlow`) and
`start_run` executes it:
```rust
let graph = DataflowGraph::new()
.with_dataset(Dataset::new("knut_nums", OutputType::MaterializedView))
.with_flow(Flow::new("knut_nums", "knut_nums", Vec::<String>::new())
.with_query("SELECT id FROM range(0, 1000)"));
let id = client.define_graph(&graph).await?; // client.rs:348 → the four commands
let run = PipelineRun { graph_id: Some(id), full_refresh_all: true,
storage: Some("file:///tmp/pipeline".into()), ..Default::default() };
let mut sink = knut_pipelines::StdoutEventSink::new(); // viewer.rs:195
client.start_run(&run, &mut sink).await?; // client.rs:367 — streams the lifecycle
```
Two field gotchas the live tests pinned (both handled): `DefineFlow.once` must be
**unset** for normal flows, and `StartRun` needs a `storage` URI
(`PipelineRun::storage`) or outputs are never materialised.
## The `--json-logs` contract
`knutpipe --json-logs …` emits the client's `knutpipe::{client,run}` tracing
events as JSON lines a test can drive — `connected`, `pipeline_event`,
`job_progress`:
```text
{"fields":{"event":"connected","url":"sc://127.0.0.1:15002"},"target":"knutpipe::client"}
{"fields":{"event":"job_progress","fraction":0.51,"tasks":25,"total":49},"target":"knutpipe::run"}
```
For the GUI, set `KNUT_VIEWER_STATE=<path>` and poll that file: it holds
`ViewerState::state_json()` (datasets, node_status, job_progress, events,
placeholders), refreshed ~every 300 ms — the headless post-condition oracle.
`KNUT_SCREENSHOT=<path>` captures one GUI frame and exits.
## Running with the facett-adapter feature
To render the same `ViewerState` through facett's engine-agnostic viewer, build
with `--features facett-adapter` and hand the state to facett:
```rust
// cargo run/test -p knut-pipelines --features facett-adapter
use facett_pipeline::{PipelineAdapter, PipelineView};
let view = PipelineView::from_adapter(&viewer_state); // ViewerState impls PipelineAdapter
// or the free function when you only want the neutral model:
let model = knut_pipelines::to_pipeline_model(&viewer_state); // -> facett PipelineModel
```
Each `Dataset` becomes a `Stage`, each `read → target` an edge, Spark
`ExecutionProgress` becomes `PipelineMetrics`. Your plain `knutpipe` build is
untouched — facett only enters under this feature. Full mapping table:
[`pipelines-design.md#f7`](pipelines-design.md). Tests:
`cargo test -p knut-pipelines --features facett-adapter`.