dataflow-spec-builder

name

dataflow-spec-builder

description

Generates Data Flow Spec pipeline bundles using the metadata-driven framework from github.com/databricks-solutions/lakeflow_framework. Creates Data Flow Specs (standard, flows, materialized views), schemas, expectations, SQL transforms, Python extensions, substitutions, templates, pipeline resource YAMLs, and databricks.yml from natural language. Covers CDC (SCD1/2), data quality, quarantine, liquid clustering, multi-source streaming, table migration, secrets, soft deletes, operational metadata, and DABs deployment. This is NOT native Lakeflow Declarative Pipelines — it is the Data Flow Spec Framework that wraps SDP with configuration-over-code.

Data Flow Spec Builder

Generate complete, production-ready pipeline bundles using the Data Flow Spec Framework (v0.4.0) from natural language descriptions. This is a metadata-driven data engineering framework that wraps Databricks Spark Declarative Pipelines (SDP) with a configuration-over-code approach using JSON/YAML Data Flow Specs.

IMPORTANT: This skill builds pipelines using the Data Flow Spec Framework (databricks-solutions/lakeflow_framework), NOT native Lakeflow Declarative Pipelines (DLT). The Data Flow Spec Framework is a metadata-driven wrapper that uses JSON configuration files called "Data Flow Specs" to define pipelines declaratively.

When to Use

• User says "generate a Data Flow Spec", "use the Data Flow Spec framework", "scaffold a DFS pipeline bundle", or "dataflow spec builder"
• User says "use the metadata-driven framework" or "databricks-solutions/lakeflow_framework"
• User needs metadata-driven SDP pipelines with JSON/YAML Data Flow Spec configuration files
• User wants CDC (SCD1/2), data quality expectations, quarantine, or multi-source streaming via Data Flow Specs
• User asks for medallion architecture pipelines using Data Flow Specs (not native DLT/Lakeflow syntax)
• User wants to create reusable pipeline templates for multiple similar tables using the Data Flow Spec Framework
• User needs environment-portable pipelines with substitutions (dev/sit/prod) using Data Flow Spec configuration
• User references databricks-solutions/lakeflow_framework or "DFS framework"

When NOT to Use

• User wants native Lakeflow Declarative Pipelines (DLT) with Python/SQL decorators like @dlt.table or CREATE STREAMING TABLE — that is native DLT, not this framework
• User wants to write pipeline code directly without JSON/YAML configuration files

Parameters

Parameter	Required	Description
description	Yes	Natural language description of the pipeline(s) to build
catalog	Yes	Unity Catalog name (e.g., main)
schema	Yes	Target schema (e.g., energy_workshop)
framework_source_path	No	Workspace path where the framework bundle is deployed (default: /Workspace/Users/{owner}/.bundle/lakeflow_framework/dev/current/files/src)
spec_format	No	json (default) or yaml
layer	No	Medallion layer: bronze, silver, or gold
pattern	No	Pipeline pattern to use (see Patterns section)
environments	No	List of target environments for substitutions (e.g., dev,sit,prod)

Workflow

1. Analyze request — Determine data flow type (standard / flows / materialized_view), pattern, layer, sources, targets, and features needed
2. Generate Data Flow Spec — Create the main JSON/YAML spec file(s) with correct schema
3. Generate schemas — Create StructType JSON schema files for sources and targets
4. Generate expectations — Create data quality expectation files if DQ is needed
5. Generate SQL transforms — Create SQL files for silver/gold transforms
6. Generate Python extensions — Create custom source/transform/sink modules if complex logic is needed
7. Generate substitutions — Create environment-specific substitution configs
8. Generate pipeline resource YAML — Create resources/*.yml with pipeline definition and filters
9. Generate databricks.yml — Create the DAB bundle definition
10. Generate templates — If the user has multiple similar tables, create reusable Lakeflow templates
11. Deploy — Optionally run databricks bundle deploy and databricks bundle run

---

Architecture Overview

The Lakeflow Framework has two bundle types:

Framework Bundle (deployed once per workspace)

Contains the core engine code. Deployed to workspace files via:

cd lakeflow_framework
databricks bundle deploy -t dev

Pipeline Bundle (one per project/domain)

Contains your Data Flow Specs, schemas, expectations, transforms, and pipeline definitions:

my_pipeline_bundle/
├── databricks.yml              # DAB bundle definition
├── resources/
│   └── my_pipeline.yml         # SDP pipeline definition(s)
└── src/
    ├── dataflows/
    │   └── my_table/
    │       ├── dataflowspec/
    │       │   └── my_table_main.json    # Data Flow Spec
    │       ├── schemas/
    │       │   └── my_table_schema.json  # StructType schema
    │       ├── expectations/
    │       │   └── my_table_dqe.json     # DQ expectations
    │       └── dml/
    │           └── transform.sql         # SQL transforms
    ├── extensions/                       # Python extensions (optional)
    │   ├── sources.py
    │   ├── transforms.py
    │   └── sinks.py
    ├── templates/                        # Reusable templates (optional)
    │   └── my_template.json
    └── pipeline_configs/
        ├── dev_substitutions.json        # Env substitutions
        └── pipeline_config.json          # Pipeline-level config

---

Data Flow Types

1. Standard Data Flow (Bronze / 1:1 Ingestion)

Best for: ingestion, basic 1:1 loads, single-source-to-single-target.

{
    "dataFlowId": "<unique_id>",
    "dataFlowGroup": "<group_name>",
    "dataFlowType": "standard",
    "sourceType": "<cloudFiles|delta|deltaJoin|kafka>",
    "sourceSystem": "<source_system_name>",
    "sourceViewName": "v_<view_name>",
    "sourceDetails": {
        "database": "<catalog.schema>",
        "table": "<source_table>",
        "cdfEnabled": false,
        "schemaPath": "schemas/<schema_file>.json",
        "selectExp": ["*"],
        "whereClause": [],
        "readerOptions": {}
    },
    "mode": "<stream|batch>",
    "targetFormat": "delta",
    "targetDetails": {
        "table": "<target_table>",
        "schemaPath": "schemas/<target_schema>.json",
        "tableProperties": {
            "delta.autoOptimize.optimizeWrite": "true",
            "delta.autoOptimize.autoCompact": "true"
        },
        "partitionColumns": [],
        "clusterByColumns": [],
        "clusterByAuto": false
    },
    "cdcSettings": {
        "keys": ["<primary_key>"],
        "sequence_by": "<timestamp_column>",
        "scd_type": "<1|2>",
        "where": "",
        "ignore_null_updates": true,
        "except_column_list": [],
        "apply_as_deletes": "",
        "track_history_column_list": [],
        "track_history_except_column_list": []
    },
    "dataQualityExpectationsEnabled": false,
    "dataQualityExpectationsPath": "",
    "quarantineMode": "<off|flag|table>",
    "quarantineTargetDetails": {
        "targetFormat": "delta",
        "table": "<quarantine_table>",
        "tableProperties": {}
    },
    "tableMigrationDetails": {
        "enabled": false,
        "catalogType": "<hms|uc>",
        "autoStartingVersionsEnabled": true,
        "sourceDetails": {
            "sourceMigrateDelta": {
                "database": "<old_database>",
                "table": "<old_table>",
                "selectExp": [],
                "whereClause": [],
                "exceptColumns": []
            }
        }
    }
}

Source Types for Standard:

• cloudFiles — Auto Loader from UC Volumes or cloud storage (S3/ADLS/GCS)
• delta — Existing Delta table, optionally with CDF
• deltaJoin — Join multiple Delta tables (stream + static)
• kafka — Apache Kafka topics

Source Details by Type:

For cloudFiles:

"sourceDetails": {
    "path": "/Volumes/<catalog>/<schema>/<volume>/<folder>/",
    "readerOptions": {
        "cloudFiles.format": "<csv|json|parquet|avro>",
        "header": "true",
        "cloudFiles.inferColumnTypes": "true",
        "cloudFiles.schemaLocation": "<checkpoint_path>"
    },
    "schemaPath": "schemas/<name>.json"
}

For delta:

"sourceDetails": {
    "database": "<catalog>.<schema>",
    "table": "<table_name>",
    "cdfEnabled": true,
    "schemaPath": "schemas/<name>.json",
    "selectExp": ["col1", "col2", "CAST(col3 AS DATE) AS col3_date"],
    "whereClause": ["col1 IS NOT NULL"]
}

For deltaJoin:

"sourceDetails": {
    "sources": [
        {
            "alias": "a",
            "database": "<catalog>.<schema>",
            "table": "<table_1>",
            "mode": "stream",
            "cdfEnabled": true
        },
        {
            "alias": "b",
            "database": "<catalog>.<schema>",
            "table": "<table_2>",
            "mode": "static"
        }
    ],
    "joinType": "<inner|left>",
    "joinCondition": "a.key_col = b.key_col",
    "selectExp": ["a.*", "b.description"]
}

For kafka:

"sourceDetails": {
    "readerOptions": {
        "kafka.bootstrap.servers": "<broker:9092>",
        "subscribe": "<topic_name>",
        "startingOffsets": "earliest"
    },
    "schemaPath": "schemas/<name>.json"
}

2. Flows Data Flow (Silver / Multi-Source Streaming)

Best for: complex transformations, multiple sources merging into one target, silver/gold layers.

{
    "dataFlowId": "<unique_id>",
    "dataFlowGroup": "<group_name>",
    "dataFlowType": "flow",
    "targetFormat": "delta",
    "targetDetails": {
        "table": "<target_table>",
        "schemaPath": "",
        "tableProperties": {
            "delta.enableChangeDataFeed": "true"
        },
        "partitionColumns": [],
        "clusterByColumns": []
    },
    "cdcSettings": {
        "keys": ["<primary_key>"],
        "sequence_by": "<timestamp_column>",
        "scd_type": "<1|2>",
        "where": "",
        "ignore_null_updates": true,
        "except_column_list": []
    },
    "dataQualityExpectationsEnabled": false,
    "quarantineMode": "off",
    "quarantineTargetDetails": {},
    "flowGroups": [
        {
            "flowGroupId": "<flow_group_id>",
            "stagingTables": {
                "<staging_table_name>": {
                    "type": "<ST|MV>",
                    "schemaPath": "",
                    "partitionColumns": [],
                    "tableProperties": {},
                    "cdcSettings": {}
                }
            },
            "flows": {
                "<flow_name>": {
                    "enabled": true,
                    "flowType": "<append_view|append_sql|merge>",
                    "flowDetails": {
                        "targetTable": "<staging_or_target_table>",
                        "sourceView": "<view_name>"
                    },
                    "views": {
                        "<view_name>": {
                            "mode": "<stream|batch>",
                            "sourceType": "<cloudFiles|delta|deltaJoin|sql|python>",
                            "sourceDetails": {
                                "database": "<catalog>.<schema>",
                                "table": "<source_table>",
                                "cdfEnabled": true,
                                "selectExp": ["*"],
                                "whereClause": []
                            }
                        }
                    }
                }
            }
        }
    ]
}

Flow Types:

• append_view — Append data from a source view to a staging/target table
• append_sql — Append data using a raw SQL statement (use sqlPath in flowDetails instead of sourceView)
• merge — CDC merge from a source view/staging table to the target table

Staging Table Types:

• ST — Streaming Table
• MV — Materialized View

Multi-Source Streaming Pattern: Stream multiple sources into a single staging table via append_view flows, then merge the staging table into the final target. Flow groups can be added/removed over time without requiring a full pipeline refresh.

3. Materialized View Data Flow (Gold / Aggregations)

Best for: precomputed aggregations, complex joins, gold layer KPIs.

{
    "dataFlowId": "<unique_id>",