Development Guide
A complete reference for pipeline designers, data engineers, and developers building on or extending DataKnits. Covers the visual canvas, all transformation types, and code generation internals.
On this page
Pipeline Canvas
The Canvas is the visual IDE where you design data flows. It is a DAG (Directed Acyclic Graph) editor with a drag-and-drop interface.
Adding a Source Node
- Open the Metadata Catalog in the left sidebar.
- Expand a connector → schema → table.
- Drag the table onto the canvas. A Source Node is created automatically at the drop position, pre-populated with connection details.
Keyboard Shortcuts
| Shortcut | Action |
|---|---|
Ctrl / ⌘ + S | Save pipeline draft |
Ctrl / ⌘ + Z | Undo |
Ctrl / ⌘ + Shift + Z | Redo |
Del / Backspace | Delete selected node |
Space + Drag | Pan canvas |
Ctrl / ⌘ + Scroll | Zoom in / out |
Ctrl / ⌘ + Shift + V | Validate pipeline |
Node Types
| Node | Category | Description |
|---|---|---|
| Source Node | I/O | Reads data from a connected technology. One per data source per pipeline branch. |
| Target Node | I/O | Writes data to a destination. Supports Append, Overwrite, Upsert, Merge write modes. |
| Filter Node | Transform | Removes rows based on WHERE-clause conditions. |
| Join Node | Transform | Inner, Left, Right, Full Outer, Cross, Semi, Anti joins with automatic broadcast hints for small tables. |
| Aggregate Node | Transform | Group By with Sum, Avg, Count, Min, Max, First, Last, Collect List, Count Distinct, having-clause support. |
| SCD Node | Transform | Slowly Changing Dimension types 1, 2, 3, and 4, plus Surrogate Key generation. |
| Derive Node | Transform | Adds or replaces columns with SQL expressions (withColumn). |
| Custom SQL Node | Transform | Arbitrary Spark SQL via createOrReplaceTempView + spark.sql(). |
| Data Quality Node | Quality | Validation rules: not_null, range, regex, custom expression. Quarantine or fail on violation. |
| Repartition Node | Performance | Hash, range, or coalesce repartitioning for downstream optimization. |
| Cache Node | Performance | Cache / persist DataFrames at configurable StorageLevels to avoid recomputation. |
39 Transformation Components
Every transformation node in the canvas maps 1:1 to a code generator (currently implemented for the PySpark engine). Components are grouped below by category, matching the codegen source layout.
Basic (Row & Column Operations)
| Component | PySpark API | Notes |
|---|---|---|
| Filter | df.filter(F.expr(...)) | Include or exclude mode; optional SQL mode for QUALIFY / LATERAL VIEW syntax. |
| Select | df.select(F.col(), F.expr().alias()) | Column projection with expression aliases. |
| Rename | withColumnRenamed chain | |
| Cast | .cast() | All standard Spark types; date/timestamp format strings supported. |
| Drop Columns | df.drop(...) | |
| Derive | withColumn(F.expr()) | Add/replace columns; optional target-type cast. |
| Sort | df.orderBy() | Nulls first / last configurable per column. |
| Deduplicate | dropDuplicates() / select().distinct() | ALL_COLUMNS or KEY_ONLY output mode. |
| Fill NA | df.fillna() | Per-column dict or single global value. |
| Drop NA | df.dropna(how, subset) | any / all mode, optional column subset. |
| Limit | df.limit(n) | Emits a warning to verify intent in production. |
| Multi-Transform Sequence | Chained withColumn() steps | Column-level pipeline of micro-transformations — see Column Function Library below. |
Advanced
| Component | PySpark API | Notes |
|---|---|---|
| Join | inner / left / right / full / cross / left_semi / left_anti | Python-level broadcast hint or SQL hint mode (BROADCAST / SHUFFLE_HASH / SHUFFLE_MERGE / SKEW). |
| Union | union() / unionByName() | Also supports INTERSECT / EXCEPT via SQL temp views. |
| Aggregate | groupBy().agg(), rollup(), cube(), GROUPING SETS | count, sum, avg, min, max, first, last, collect_list, collect_set, countDistinct, approx_count_distinct, percentile_approx; optional HAVING clause. |
| Window | Window.partitionBy().orderBy() | row_number, rank, dense_rank, percent_rank, cume_dist, lag, lead, ntile, sum, avg, min, max, count, first, last; per-function frame overrides (rowsBetween / rangeBetween). |
Utility
| Component | PySpark API | Notes |
|---|---|---|
| Sample | df.sample(withReplacement, fraction, seed) | |
| Lookup | Broadcast join against a reference dataset | Selects only needed lookup columns; optional cache; fills defaults for unmatched rows. |
| Custom UDF | Python, SQL, Pandas UDF, or Scala | 5 Pandas UDF variants: series_to_series, iterator_series, series_to_scalar, map_in_pandas, apply_in_pandas. |
| Add Audit Columns | withColumn() | Injects _load_ts, _run_id, _run_user (configurable names). |
| Case When | F.when().when().otherwise() | Single output column, ordered condition list. |
| Broadcast Variable | sparkContext.broadcast() | Optional UDF-based lookup columns generated from the broadcast dict. |
| Accumulator | sparkContext.accumulator() | Driver-side counter across executors; int or float. |
Special
| Component | PySpark API | Notes |
|---|---|---|
| Pivot | groupBy().pivot().agg() | Optional explicit pivot value list. |
| Unpivot | F.expr("stack(...)") | |
| Explode | F.explode() / F.explode_outer() | |
| Flatten | Inline _flatten_df() helper | Flattens nested struct columns; configurable separator and column subset. |
| Custom SQL | createOrReplaceTempView + spark.sql() | Multi-input support via {input} / {input1} placeholders. |
| Data Quality | Filter/count-based rule checks | not_null, range, regex, unique, custom rules; fail, drop, quarantine, or warn-only. |
| Repartition | repartition() / repartitionByRange() / coalesce() | Hash, range, or coalesce strategy. |
| Cache | df.cache() / df.persist(StorageLevel.X) | Optional eager materialization via .count(). |
| Iceberg DML | spark.sql("DELETE/UPDATE/MERGE INTO ...") | Row-level DML on Iceberg tables. WHERE clause mandatory for DELETE/UPDATE. Pass-through node — does not modify the upstream DataFrame. |
| Data Profile | describe(), null counts, approx_count_distinct() | Read-only, pass-through — logs profile stats without modifying data. |
| Mask | hash / null / truncate / replace / regex_replace | PII masking; hash uses F.sha2(..., 256). |
SCD & Keys
| Component | PySpark API | Notes |
|---|---|---|
| SCD Type 1 | Delta MERGE INTO | Overwrites matched records, inserts new ones. No history. |
| SCD Type 2 | Hash-based change detection + Delta/Iceberg MERGE | Closes changed records, inserts new versions. Optional soft-delete handling closes rows absent from source. |
| SCD Type 3 | Join + column shift | Maintains prev_{column} alongside current value for tracked columns. |
| SCD Type 4 | Hash-based change detection + separate history table write | Main table keeps current row only; changed rows are appended to a dedicated history table (Delta or Iceberg). |
| Surrogate Key | monotonically_increasing_id() / uuid() / row_number() | row_number strategy requires orderBy for deterministic keys. |
Column Function Library (Multi-Transform Sequence)
The Multi-Transform Sequence component chains column-level micro-transformations, each compiled to a PySpark column expression. The library covers 169 functions across 14 categories. Each category links to a dedicated reference page with full syntax, parameters, and defaults for every function.
| Category | Count | Description |
|---|---|---|
| Convert | 3 | Type coercion between numbers, dates, and target types. |
| String | 25 | Trimming, casing, padding, substring, and character-level operations. |
| Date / Time | 44 | Date arithmetic, part extraction, formatting, and timezone handling. |
| Regex | 5 | Pattern extraction, replacement, and matching. |
| Conditional / Null Handling | 7 | Null coalescing, CASE WHEN logic, and list membership checks. |
| Arithmetic | 4 | Add, subtract, multiply, divide against a second operand. |
| Numeric | 42 | Rounding, trigonometry, logarithms, random numbers, and bitwise ops. |
| Type / Encoding | 5 | Hex/binary/base conversions, safe casting, and custom SQL expressions. |
| String Extras / Hashing | 16 | Splitting, formatting, hashing (SHA1/MD5), and URL/Base64 encoding. |
| Crypto | 2 | AES encryption and decryption. |
| JSON | 2 | Extracting values from and parsing JSON strings. |
| Array | 7 | Array containment, sizing, sorting, and joining. |
| Struct / Map | 3 | Reading struct fields and map keys/values. |
| Boolean Aggregates | 4 | every, some, count_if, any_value. |
Each step supports an onError policy (RETURN_NULL or USE_DEFAULT with a configured default value), and steps within a sequence chain on the output of the previous step.
Slowly Changing Dimensions (SCD)
The SCD node handles historical tracking for dimension tables without writing custom merge logic. See the SCD & Keys table above for the full list of implemented types (1, 2, 3, 4) and the Surrogate Key generator.
| Type | Strategy | Columns Added / Written |
|---|---|---|
| SCD Type 1 | Overwrite old values with new values. No history retained. | None |
| SCD Type 2 | Insert new row for changes. Marks old row as inactive. Full historical tracking. Optional soft-delete handling closes rows absent from source. | end_date, is_current-style flag column (configurable names) |
| SCD Type 3 | Partial history. Stores current and previous values in separate columns. | prev_{col} for each tracked column |
| SCD Type 4 | Main table keeps only the current row; changed rows are written to a separate history table. | effective_date column on history table rows |
Pushdown Eligibility System
DataKnits automatically analyzes each pipeline segment and surfaces a real-time eligibility indicator. You can toggle execution points between Source DB and PySpark from the canvas.
Eligibility Rules
- Eligible for Source DB pushdown — All functions in the segment are natively supported by the source engine, no PySpark-derived column references, single source only.
- Forced to PySpark — Cross-source JOINs, derived column lineage breaks, or functions not supported in source.
Function Availability Badges
| Badge | Meaning |
|---|---|
| ✓ Available | Works natively in the source database. |
| ⚠ Alternative | Works with different syntax (e.g., CASE WHEN instead of IF() in Oracle). DataKnits generates the correct dialect automatically. |
| ✗ Unavailable | Requires PySpark. Canvas offers a one-click switch to PySpark execution for that segment. |
Execution & Monitoring
After clicking Run, the Execution Monitor shows real-time progress of your pipeline:
- Live node-level status (queued, running, success, failed) on the canvas.
- Structured log stream with log-level filtering.
- Row counts in / out for each node.
- Execution time per node and total wall-clock time.
- Failed node inspection with error message and stack trace.
Re-running Failed Jobs
Operators can re-run any failed job within the last 30 days from the execution history. The original pipeline version and environment config are used — changes to the pipeline draft do not affect the re-run.