Your browser sucks, JavaScript disabled! Gunnar's console needs JavaScript and WebAssembly. Git over SSH on port 2222 does not.

nordisk/knut

crates/knut-pipelines/docs/pipelines-viewer.md

knut-pipelines: a viewer for Spark Declarative Pipelines

knut-pipelines connects to a Spark Connect server, fetches the compiled dataflow DAG of a Spark Declarative Pipeline (SDP) and the run events it emits, and renders both — building toward a live stream viewer where events appear as a run progresses.

How to try it & view it

All builds go through CARGO_TARGET_DIR=/mnt/work4t/cargo-target/knut (never the flaky T9 source disk). Commands below assume that env is set.

Static (default build — no server, no protoc, no GUI): renders a graph as a text outline (with topo order) or JSON. With no path it renders a built-in bronze → silver → gold demo; give it a spec (see examples/pipeline.json) to render your own. Flow edges are completed by walking each flow's SQL query, so a spec that carries only queries still shows full edges.

cargo run -p knut-pipelines --bin knutpipe -- show                    # demo graph
cargo run -p knut-pipelines --bin knutpipe -- show path/to/graph.json
cargo run -p knut-pipelines --bin knutpipe -- show --json             # DAG as JSON

Live (grpc feature): connect to a Spark Connect server and stream a pipeline run — or any Spark SQL job's progress — to the terminal. Uses protoc-bin-vendored, so no system protoc is needed.

cargo run -p knut-pipelines --features grpc --bin knutpipe -- \
    watch sc://HOST:15002 --graph GRAPH_ID          # stream an existing run
cargo run -p knut-pipelines --features grpc --bin knutpipe -- \
    watch sc://HOST:15002 --sql "SELECT count(*) FROM range(1e9)"   # a SQL job
cargo run -p knut-pipelines --features grpc --bin knutpipe -- \
    pipeline sc://HOST:15002 --spec examples/pipeline.json          # define + run

This REQUIRES a running Spark 4.x Connect server with spark.pipelines (SDP GA'd in Apache Spark 4.1.0). There is no way to fake the cluster; without a reachable server these commands return an honest transport error.

Desktop GUI (gui, or live = gui + grpc): an egui + egui-snarl node graph coloured by live status, with job-progress bars and an event log.

cargo run -p knut-pipelines --features gui  --bin knutpipe -- gui           # static explorer
cargo run -p knut-pipelines --features live --bin knutpipe -- gui --connect sc://HOST:15002

Web viewer (web feature): an axum + SSE server that serves facett's Rust facett-pipeline DAG viewer compiled to wasm (the facet-wrapped bundle, from --facet-dist DIR / FACET_WRAPPED_DIST) and streams it PipelineModel frames (to_pipeline_model encoded by facett_pipeline::wire). knut draws nothing in JavaScript. Demoable with no Spark server — it serves the built-in demo graph and replays a synthetic run over SSE so the DAG animates; open the printed URL.

cargo run -p knut-pipelines --features web       --bin knutpipe -- web            # no-server demo (http://127.0.0.1:8787)
cargo run -p knut-pipelines --features web       --bin knutpipe -- web graph.json --addr 127.0.0.1:9000
cargo run -p knut-pipelines --features "web grpc" --bin knutpipe -- web --connect sc://HOST:15002   # live run

The live --connect form needs grpc and a running Spark 4.x Connect server. Events are driven through the same PipelineEventSink + ViewerState as the CLI/GUI (the page renders ViewerState::state_json snapshots) — the web view shows nothing the show/dump_json CLI can't.

Airflow asset events (airflow feature): knut_pipelines::AirflowAssetSink is a PipelineEventSink — drive a run's events through it and each completed SDP output becomes an AIP-82 asset event, emitted via any knut_bifrost::airflow::AssetEmitter (e.g. bifröst's HTTP AirflowEmitter). Library API, no dedicated CLI verb; see the sink's unit tests for a synthetic run.

What's missing

Honest gaps as of this writing: real runs need a live Spark 4.x Connect server with spark.pipelines (SDP is Spark 4.1.0+) — the static show view, the web demo and all unit tests run without one, but watch/pipeline/live-gui/live-web do not, so you cannot see a real run on this box. On the sibling engine, knut-thund native streaming execution is deferred (backend/native.rs): the batch dataflow runs on DataFusion, but unbounded/live sources are reported as deferred rather than faked.

What is SDP?

Spark Declarative Pipelines (spark.pipelines, SPARK-51727, Spark 4.x) is the open-sourced engine behind Databricks DLT. You declare outputs and the queries that populate them; the engine compiles a dataflow graph, works out execution order, and runs it — incrementally for streaming sources, fully for materialized views.

The wire protocol is in spark/connect/pipelines.proto:

How the commands ride Spark Connect

There is no dedicated pipelines RPC. Everything flows through the existing SparkConnectService.ExecutePlan server-streaming RPC:

ExecutePlanRequest
  └─ Plan.op_type = Command
       └─ Command.command_type = PipelineCommand (commands.proto field 19)
            └─ one of: CreateDataflowGraph | DefineOutput | DefineFlow
                       | StartRun | DefineSqlGraphElements | ...

ExecutePlanResponse           (server stream)
  ├─ pipeline_command_result  (base.proto field 22)  -> graph id, resolved ids
  └─ pipeline_event_result    (base.proto field 21)  -> PipelineEvent (the live stream)

So StartRun is sent as one ExecutePlan call and the events arrive as a stream of responses on that same call — exactly the shape a live viewer wants.

Crate architecture

src/
  error.rs    PipelineError + Result (thiserror), workspace idiom
  model.rs    DEFAULT BUILD — hand-written, dependency-light DAG + events:
                DataflowGraph { datasets: Vec<Dataset>, flows: Vec<Flow>, ... }
                Dataset { name, output_type, comment, format, ... }
                Flow    { name, target, reads, once, source_location }
                PipelineRun, PipelineRunEvent, EventLevel, OutputType
              + validate(), topo_order() (Kahn, cycle-detecting), flows_into()
  viewer.rs   DEFAULT BUILD — rendering:
                dump_text(&graph) -> String   (indented node/edge outline + topo order)
                dump_json(&graph) -> String   (serde_json pretty)
                trait PipelineEventSink { on_event, on_finish }  <- the live seam
                StdoutEventSink<W: Write>      (the seed live viewer)
  client.rs   FEATURE `grpc` — Spark Connect gRPC client (PipelinesClient):
                connect(sc://host:port) ; create_dataflow_graph ;
                define_output / define_flow / define_graph ;
                start_run(&run, &mut sink)        -> push events into a sink
                run_event_stream(&run)            -> Stream<PipelineRunEvent>
              + private `proto` module = tonic::include_proto!("spark.connect")
proto/spark/connect/*.proto   vendored Spark Connect protos (11 files)
build.rs                      FEATURE `grpc` — tonic-prost-build codegen

The model is the contract, not the generated code

The default build compiles with no protoc and no network: model and viewer are plain Rust. The generated protobuf types live only behind the grpc feature, and the client converts them into the model types at the boundary. Frontends (text dump, JSON, a future egui/ratatui/web UI) depend on model + viewer and never see a generated struct. This mirrors how knut-bifrost keeps its event model dependency-free and its Spark/Airflow transports behind features.

Proto generation approach

The full spark/connect/*.proto set is interdependent: pipelines.proto imports common, expressions, relations, types; relations pulls in catalog, ml, ml_common; base/commands tie it together. Rather than chase a partial transitive closure, we vendor the entire (self-contained) set — 11 files, only depending on the google.protobuf well-known types:

base.proto  catalog.proto  commands.proto  common.proto  example_plugins.proto
expressions.proto  ml.proto  ml_common.proto  pipelines.proto  relations.proto
types.proto

They were fetched from apache/spark@master (sql/connect/common/src/main/protobuf/spark/connect/). Note: at the time of writing the v4.0.0 tag did not carry pipelines.proto (it landed after 4.0); the vendored copy is the 4.2-era proto (it has the *_identifier / ExecuteOutputFlows additions). When pinning to a released Spark, re-vendor from that exact tag.

build.rs (only when grpc is on):

  1. prefers a system protoc, falling back to the protoc-bin-vendored binary so the feature builds on machines without protoc;
  2. compiles the 11 protos with tonic-prost-build::configure() (tonic 0.14 moved prost codegen out of tonic-build into tonic-prost-build);
  3. emits client stubs only (build_server(false)).

tonic version alignment

The workspace is now on tonic 0.14 / arrow 59 (Cargo.toml: tonic = "0.14", arrow = "58"). This crate's gRPC stack pins the same tonic 0.14 line, matching the proto vintage and the tonic tree spark-connect carries — so there is no cross-version bridge to manage here. It is still confined to the off-by-default grpc feature (its generated protobuf types never cross into the rest of the workspace), so cargo build --workspace on default features pulls in neither protoc nor the gRPC tree and stays green.

Roadmap toward the live viewer

Hard requirement: CLI parity, never GUI-only. Every capability the TUI/egui/web viewer gains must also be reachable from the command line (dump_text/dump_json + a streaming stdout sink are the floor). A GUI is an additional frontend over the same model + PipelineEventSink, never the only way to see a pipeline.

  1. Done — model + static viewer + gRPC seam. Define a graph, start a run, stream PipelineRunEvents into a PipelineEventSink; StdoutEventSink prints them line by line.
  2. Done — unified viewer state + Spark job progress. progress::ViewerState is the single model both the CLI and the GUI render: the static graph, per-dataset NodeStatus derived from the event stream, the event log, and the latest Spark ExecutionProgress (per-stage task counts + bytes, mirror of base.proto). client::execute_sql streams that progress for any Spark Connect SQL — so the viewer is a Spark-jobs monitor, not only a pipeline DAG. run_updates merges pipeline events and job progress into one ViewerUpdate stream.
  3. Done — knutpipe CLI. knutpipe show [graph.json] [--json] (static, default features), knutpipe watch <sc-url> [--graph ID | --sql ...] (live, grpc), knutpipe gui [graph.json] [--connect sc://…] (gui). viewer::dump_state is the CLI twin of the GUI — the parity floor.
  4. Done — egui + egui-snarl desktop viewer (gui feature). gui::launch runs an eframe app: the DAG is an egui-snarl node graph laid out in topological layers, nodes/pins/wires coloured by live NodeStatus (idle/running/done/failed); a side panel shows Spark job-progress bars (overall + per stage) and a colour-coded event log. With grpc (the live feature), a background thread drives run_updates and streams into the view over an mpsc channel; without it, it's a static graph explorer. Version note: egui-snarl 0.4 targets egui 0.28, so the whole egui/eframe stack is pinned to 0.28 for one consistent, fully-cached build. Bump all three together when moving to a newer snarl.
  5. Done — complete edges + richer events. DataflowGraph::populate_reads() walks each flow's defining relation (its SQL query) and fills Flow::reads with the in-graph datasets it references — so a spec that carries only queries renders full edges + a real topo order without replaying the definitions against a server (wired into knutpipe show/pipeline). progress::parse_event_message() recovers the structure the DAG needs from the free-text PipelineEvent message (the proto still carries only a timestamp + message): a quoted element, a flow status keyword (STARTING/COMPLETED/FAILED…), and a wrote N rows count. ViewerState now attributes an event to a node by flow name → target dataset (not just a dataset scan), sets status from the parsed keyword, and records rows_written per dataset (surfaced in state_json + dump_state); the gRPC event_from_proto sets the event element so the live wire path carries attribution too.
  6. Done — web viewer (web feature). web::serve is an axum + SSE server: GET /graph.json reuses dump_json; GET /events streams PipelineModel frames (to_pipeline_model encoded by facett_pipeline::wire, one per run event) driven through a WebEventSink (the PipelineEventSink seam — no duplicated model); GET / sends the browser to facett's facet-wrapped loader, which mounts the Rust facett-pipeline viewer compiled to wasm and feeds it the frames. knut draws nothing in JavaScript (lane SW row 6, 2026-09-05, deleted the earlier SVG/JS drawing). The bundle is served from --facet-dist / FACET_WRAPPED_DIST; a missing bundle is a 503 naming the files, never a blank page. knutpipe web demos it with no Spark server — the built-in demo graph + a synthetic replay animate the DAG; --connect (with grpc) streams a live run. The CLI show/dump_json stays the parity floor.
  7. Done — Bifröst hook (AIP-82). airflow::AirflowAssetSink (feature airflow) is a PipelineEventSink that forwards each SDP output completing a run as an Airflow asset event via knut_bifrost::airflow::AssetEmitter, so a downstream Airflow DAG can schedule on a pipeline output. It reuses bifröst's AssetEvent/AssetEmitter and this crate's ViewerState (no duplicated model or parsing): each dataset emits one asset event (URI + rows + output type) the first time it reaches Done; a failed output emits nothing. Collection is synchronous (the sink seam); emit is a separate async step through any AssetEmitter.

Autonomy decisions (2026-06-10, taken without asking, user away)

The user granted "YES on all" and left; these calls were made to keep moving: