dataflow-spec-builder

# Data Flow Spec Builder Generate complete, production-ready pipeline bundles using the [Data Flow Spec Framework](https://github.com/databricks-solutions/lakeflow_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: ```bash 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. ```json { "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`: ```json "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`: ```json "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`: ```json "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`: ```json "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. ```json { "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. ```json { "dataFlowId": "<unique_id>",