nordisk/knut

knut-pipelines — guide

Class: guide · audience: a techie driving knutpipe or the PipelinesClient API. The what is in pipelines-design.md; the why in pipelines-reasoning.md. This file is the workflow.

The knutpipe CLI

Ships src/bin/knutpipe.rs. Four subcommands; --json-logs is global.

# 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

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 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:

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:

{"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:

// 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. Tests: cargo test -p knut-pipelines --features facett-adapter.