.nornir/for-idiots.md

knut — for idiots (f0, zero to a queried graph)

Class: idiot. Every step, nothing assumed. Goal: from a cold shell to a graph you loaded, queried as SQL, and watched fill — copy-paste, no prior knut knowledge, no extra tooling. If you already know Rust and containers, read guide.md instead; this one spells out every command and what you should see.

What you need first

A terminal (the black window where you type commands) and the knut repo on disk. Type this and press Enter:

cd /run/media/rickard/T9/git/knut

The Rust toolchain (rustup) and podman for the container steps. Start podman once:

systemctl --user start podman.socket

"Redis" here means Valkey 9+ — the open-source, Redis-API/RESP-conformant store FalkorDB runs on. redis://… is just the wire-protocol scheme, not a vendor.
Every --ignored test starts and stops its own container.

Step 0 — build everything

Type each line, pressing Enter after each:

cargo build --workspace                        # all crates
cargo test  --workspace                         # all unit + doc tests (no containers)
cargo build -p knut-bifrost   --features cli  --bin bifrost
cargo build -p knut-pipelines --features grpc --bin knutpipe   # live Spark
cargo build -p knut-pipelines --features gui  --bin knutpipe   # desktop viewer

Binaries land in target/debug/. Below we call them as ./target/debug/bifrost and ./target/debug/knutpipe.

Step 1 — build the container images (once)

podman build -t knut-falkordb-valkey9 -f containers/falkordb-valkey9/Containerfile .
podman build -t knut-spark-iceberg    -f containers/spark-iceberg/Containerfile .

Prefer the public FalkorDB image? Every step honours KNUT_FALKORDB_IMAGE:

export KNUT_FALKORDB_IMAGE=localhost/knut-falkordb-valkey9:latest

Step 2 — file → graph, with NO database

bifrost gen-sample writes a sample parquet drop; --dry-run ingests it with a counting sink (no DB), printing one JSON log line per lifecycle stage.

mkdir -p /tmp/knut
./target/debug/bifrost gen-sample --files /tmp/knut --table people --rows 1000

cat > /tmp/knut/mapping.yml <<'EOF'
version: 1
tables:
  - table: people
    mode: incremental
    pipeline:
      - vertex: { label: Person, id_column: id, properties: [name, score] }
EOF

./target/debug/bifrost --json-logs run \
  --mapping /tmp/knut/mapping.yml --files /tmp/knut --dry-run --once

You should see lines with "event":"file_detected" (rows 1000) → "files_read" → "dry_run_mutation" (vertices 1000) → "table_synced" → "round_complete". Drop another file and re-run — only the new one is read (that is incremental mode):

./target/debug/bifrost gen-sample --files /tmp/knut --table people --rows 500 --chunk chunk1.parquet
./target/debug/bifrost --json-logs run --mapping /tmp/knut/mapping.yml --files /tmp/knut --dry-run --once
# → file_detected for chunk1.parquet only

The mapping YAML and every flag: bifrost-guide.md.

Step 3 — into a REAL FalkorDB (Valkey 9)

podman run -d --name knut-falkor -p 6379:6379 knut-falkordb-valkey9
sleep 3
./target/debug/bifrost --json-logs run \
  --mapping /tmp/knut/mapping.yml --files /tmp/knut \
  --falkor redis://127.0.0.1:6379 --graph demo --once
podman exec knut-falkor redis-cli GRAPH.QUERY demo "MATCH (n:Person) RETURN count(n)"

You should see a "sink_applied" event, then the count (1500). Leave the container running for the next steps, or podman rm -f knut-falkor to clean up.

Bulk-load a big graph (knut-native, no FalkorDB importer)

The population example generates a synthetic 12-label / 17-edge graph as Arrow and bulk-loads it through knut's own FalkorDbLoader (parallel UNWIND-batched Cypher) — no falkordb_bulk_loader:

cargo run -p graphar-falkordb --release --example population -- \
  --url redis://127.0.0.1:6379 --graph pop --scale 0.005 --parallel 12 --drop

You should see per-phase throughput, e.g. vertices: 606000 … 25k nodes/s (the online write ceiling — FalkorDB serializes writes per graph). --scale 1.0 ≈ 120M nodes. The four bulk methods knut has: falkordb-design.md.

Watch it fill in a GUI (`knut-popviz`)

cargo run -p knut-popviz --features gui -- --url redis://127.0.0.1:6379 --graph pop --scale 0.001 --drop

A progress bar per label + live throughput. Like the pipeline viewer it's robot-testable: --autostart + KNUT_POPVIZ_STATE=/tmp/p.json runs the load headlessly and rewrites that JSON (phase, per-label loaded/total, rows_per_s) every ~300 ms, so a test asserts what's on screen with no display. See popviz-guide.md.

Step 4 — shared state across instances (Valkey ledger)

The default ledger is a file; --state-valkey puts it in Valkey so several bifröst instances share progress and a restart resumes cleanly.

./target/debug/bifrost --json-logs run \
  --mapping /tmp/knut/mapping.yml --files /tmp/knut \
  --falkor redis://127.0.0.1:6379 --graph demo \
  --state-valkey redis://127.0.0.1:6379 --state-key bifrost:demo --once
# start log shows "state":"valkey"; a second run sees the files already done → idle

For a long-running watcher, drop --once and add --interval 5; Ctrl-C shuts down gracefully (finishes the round + persists, then exits).

Step 5 — announce to Airflow (AIP-82)

--emit-airflow POSTs one asset event per ingested table to Airflow's /api/v2/assets/events, closing the loop to downstream DAGs:

./target/debug/bifrost gen-sample --files /tmp/knut --table people --rows 10 --chunk chunk2.parquet
./target/debug/bifrost --json-logs run \
  --mapping /tmp/knut/mapping.yml --files /tmp/knut --dry-run --once \
  --emit-airflow http://airflow:8080 --asset-prefix "graph://demo/"
# → an "assets_emitted" event

Step 6 — query the graph as SQL (Power BI / ADBC), automated

The full Power-BI path (Basic handshake → bearer → GetCatalogs/GetTables → SELECT * FROM <named query>) is verified end-to-end by one test:

cargo test -p graphar-flight --test integration_flight_sql -- --ignored --nocapture

You should see test power_bi_path… ok — it starts FalkorDB, runs the Cypher Flight SQL server, and drives it with the stock FlightSqlServiceClient. How to run the server yourself: flight-guide.md.

Step 7 — graph → lakehouse (export back to GraphAr/parquet)

Run a Cypher query and write its rows as a GraphAr file (uses the FalkorDB from step 3):

cargo run -p graphar-flight --example export -- \
  redis://127.0.0.1:6379 demo \
  "MATCH (n:Person) RETURN n.id AS id, n.name AS name" \
  /tmp/knut/people.parquet
# → "exported N rows × 2 cols → /tmp/knut/people.parquet"

Straight into an Iceberg table (feature skade, the embedded Iceberg stack):

cargo run -p graphar-flight --features skade --example export-iceberg -- \
  redis://127.0.0.1:6379 demo \
  "MATCH (n:Person) RETURN n.id AS id, n.name AS name" \
  /tmp/knut_warehouse people
cargo test -p graphar-flight --features skade --lib append_and_read_back   # no container

Step 7b — Iceberg → FalkorDB delta streaming (incremental, no Spark)

The reverse of step 7: stream an Iceberg table's new rows INTO FalkorDB — and only the new rows. When the table grows by a snapshot, Bifröst reads just the data files that snapshot added (a manifest-level diff, no full rescan, no Spark) and MERGEs them, then advances a per-table cursor. Why this is hand-built: bifrost-reasoning.md.

Prove it without a database (a real local Iceberg warehouse, no container):

cargo test -p knut-bifrost --features skade --lib iceberg::tests
#   delta_reads_only_appended_rows  — A(3)+B(2): A→B delta = just {4,5}

End-to-end into real FalkorDB (--ignored container test):

cargo test -p knut-bifrost --features skade --test delta_iceberg_falkordb -- --ignored --nocapture

Step 8 — the Spark pipeline viewer (`knutpipe`)

Static (no server):

./target/debug/knutpipe show          # built-in medallion demo DAG (text)
./target/debug/knutpipe show --json    # the same, as JSON

Watch what it shows, headlessly (robot-testing — no display):

KNUT_VIEWER_STATE=/tmp/knut_viewer_state.json \
  cargo run -p knut-pipelines --features gui --bin knutpipe -- gui &
sleep 1; python3 -m json.tool /tmp/knut_viewer_state.json   # full UI state as JSON

Live against real Spark Connect:

podman run -d --name knut-spark -p 15002:15002 knut-spark-iceberg
sleep 25   # Spark's JVM takes ~10-25s to accept connections
cargo run -p knut-pipelines --features grpc --bin knutpipe -- \
  --json-logs watch sc://127.0.0.1:15002 \
  --sql "SELECT count(*) FROM range(0, 50000000, 1, 24)"
podman rm -f knut-spark

You should see "event":"connected", then climbing "event":"job_progress" lines, then the final DAG + per-stage render. Running a real Spark Declarative Pipeline and every flag: pipelines-for-idiots.md.

Step 8b — run a Þund pipeline over a local CSV (no Spark, no database)

Þund (knut-thund) is the streaming/batch engine. The shortest real run needs nothing but a file on disk: it writes a sample orders.csv, rolls it up per customer on the native Arrow/DataFusion engine, and prints the table.

cargo run -p knut-thund --example batch_file_native --features native

You should see the registered-source + materialised by_customer (2 row(s)) events, then:

+----------+-------+
| customer | total |
+----------+-------+
| bob      | 12    |
| alice    | 33    |
+----------+-------+

Point it at your own file by calling NativeBackend::with_file_input(table, path, "csv"|"parquet") — see thund-guide.md. To keep the result on disk you can either call .with_file_output("by_customer", "out/rollup.parquet", "parquet"), or just say it in the pipeline itself — Dataset::new("by_customer", OutputType::Sink).with_path("out/rollup.parquet") — so a .ron spec carries where its output lands. Asserted container-free in tests/native_file_source.rs (CSV + parquet) and tests/native_file_sink.rs (the declarative_* sink tests):

cargo test -p knut-thund --features native --test native_file_source

Step 8c — keep the result on disk (the file sink)

Step 8b gave you the answer in memory. To keep it as a file the next tool can open, add one line — .with_file_output(dataset, path, "csv"|"parquet") — and the result lands on disk. The batch_file_native example now does exactly this: it writes orders.csv, rolls it up, and writes the rollup back out as a parquet file, printing where it landed.

cargo run -p knut-thund --example batch_file_native --features native

You should also see, after the table:

wrote 2 row(s) -> /tmp/knut-thund-batch-file-native/rollup.parquet (parquet)

That is the whole round trip on a laptop: read a file → run SQL → write a file, no Spark, no database, no catalog. A sink pointed at a dataset the pipeline never produces writes no file (it doesn't fake an empty one). Asserted container-free — read → SQL → write → re-read → identical rows, for both csv and parquet:

cargo test -p knut-thund --features native --test native_file_sink

Step 8d — read only the columns and rows you need (pushdown)

If your file has 40 columns but you only want two, don't read all 40. Instead of writing SQL, tell the flow the columns and the filter and Þund reads only what you asked for straight off the disk:

Flow::batch("pick", "big", ["orders"])
    .with_filter("amount > 100")             // keep only these rows
    .with_projection(["customer", "amount"]) // keep only these columns

Þund turns that into a scan that reads only customer and amount, of only the amount > 100 rows — the other columns are never touched and non-matching chunks are skipped. You can even prove it: run.pushdown("pick") hands back the engine's plan, which shows projection=[customer, amount] and predicate=amount > 100 right on the file scan. The same flow also runs on Spark (it writes the matching SELECT … WHERE … for you). Asserted container-free:

cargo test -p knut-thund --features native --test native_projection_pushdown

Step 8e — replace vs pile up on a re-run (overwrite vs append)

Run the same pipeline twice into the same output directory and — by default — you get both runs' rows piled up (two copies of everything). That is fine for adding new data, but for a full refresh you want the second run to replace the first. Say so on the sink dataset:

Dataset::new("by_region", OutputType::Sink)
    .with_path("out/by_region")
    .with_partition_cols(["region"])
    .overwrite();   // ← wipe the folder first, so a re-run keeps ONE copy

With .overwrite() Þund clears the target folder before writing, so two runs leave one generation's rows. Without it (the default) two runs leave two — nothing about your existing pipelines changes. Prove it container-free (two runs → 3 rows overwrite, 6 rows append):

cargo test -p knut-thund --features native --test native_file_sink write_mode

Step 8f — write the pipeline as SQL instead (no Rust)

Everything above built the pipeline in Rust. If you would rather just write SQL, you can — it produces the same pipeline object, so nothing else changes:

CREATE MATERIALIZED VIEW bronze_events (
  event_id BIGINT NOT NULL,
  event_ts TIMESTAMP,
  amount   DOUBLE,
  CONSTRAINT amount_non_negative EXPECT (amount >= 0) ON VIOLATION DROP ROW
)
PARTITIONED BY (event_month)
COMMENT 'raw events'
AS SELECT event_id, event_ts, amount FROM landing_events;

CREATE MATERIALIZED VIEW silver_events AS
SELECT * FROM bronze_events WHERE amount > 0;

CREATE STREAMING TABLE gold_live AS
SELECT * FROM STREAM(silver_events);

let p = knut_thund::authoring::sql::from_sql("medallion", script)?;

You do not have to say what depends on what. Þund reads the FROM clauses and works the order out — bronze → silver → gold. Statement order in the file does not matter either.

STREAM(silver_events) is how you say "read this one as a live stream instead of a batch". That is the only word in the whole grammar that is not ordinary SQL.

The ways of writing a pipeline are interchangeable, because underneath they are the same thing:

cargo run -p knut-thund --features sql,dsl --release --example medallion_sql -- 500

That builds 1501 datasets from generated SQL, converts SQL → pipeline → RON → pipeline → SQL, checks you got back exactly what you started with, and prints the timings (about 36 ms for those 1501 datasets).

One thing that will bite you. This is legal SQL and Þund parses it happily:

PARTITIONED BY (months(event_ts))     -- ✗ nothing can actually run this

months(...) is an Iceberg partition transform. Neither Spark nor Þund's own engine can execute one, so you get a clear refusal before anything runs:

backend `spark-connect-sdp` does not support: partition transform
`months(event_ts)` on dataset `bronze_events` (this backend takes identity
partition columns only — materialize the value as a column and partition by that)

Do exactly what it says — compute the month as a real column, then partition on that column:

CAST(DATE_TRUNC('MONTH', event_ts) AS DATE) AS event_month
...
PARTITIONED BY (event_month)          -- ✓ same layout on disk

Step 9 — the full test matrix

cargo test --workspace                                   # everything container-free
cargo test --workspace --features cli -- --ignored       # + the live container e2e

Step 10 — benchmarks

cargo run -p xtask --release --example nornir-bench       # one BenchRun JSON line
cargo run -p xtask --release --example nornir-bench --features skade -- knut.skade_iceberg_export

Eleven benchers by default plus the skade Iceberg-export bencher — see benching-benchmarks.md.

Clean up

podman rm -f knut-falkor knut-spark 2>/dev/null; rm -rf /tmp/knut

Deeper than this? Every function has a design + guide chapter — start at design.md (the map). Open work lives in plan.md.