satisfactoscript 1.4.1

Declarative data engineering framework for Spark and Databricks Lakehouse.

SatisfactoScript Framework (v1.4.0)

An Enterprise-Ready, Declarative Data Engineering Framework for Spark and Databricks Lakehouse.

SatisfactoScript transforms complex PySpark pipelines into standardized, readable, and maintainable declarative contracts. By strictly decoupling the What (YAML schemas) from the How (Python business rules), it enables robust Bronze → Silver → Gold pipelines on Databricks and local Spark, optimized for Power BI Direct Query.

---

Key Capabilities

• YAML Pipeline Schemas — define sources, joins, transformations, and quality checks in readable YAML files. No more 1,000-line PySpark notebooks.
• External Declarative Sources — read CSV, Parquet, JSON, Avro, ORC, Delta, or Text files from ADLS Gen2, DBFS, S3-compatible mounts, or local storage directly from YAML — zero Python required.
• Self-documenting operators — region:equals:EMEA, amount:greater_than_equal:100, status:in:ACTIVE,PENDING. No need to memorize abbreviations.
• Smart Sandbox — in interactive mode, source tables are auto-resolved to your personal sandbox schema. Missing tables are transparently cloned from the main schema.
• Business Logic Isolation — register pure Python/PySpark rules with @RuleRegistry.register_rule().
• Semantic Layer — auto-generate semantic YAML models from your Gold tables via LLM, then query them in natural language with GenBIAgent.
• Local Spark Development Mode — run the full framework locally with local[*] PySpark + Delta Lake. No Databricks cluster required.
• Environment Aware — auto-detects Dev / QA / Prod Databricks catalogs at runtime with per-user sandbox isolation.
• Direct Query Optimized — pre-calculate OBT, YoY shifts, and distinct counts in the Gold layer to keep Power BI DAX ultra-light.

---

Architecture

Bronze (Raw)  →  Silver (Standardized)  →  Gold (Semantic / OBT)
                          │
        ┌─────────────────┴────────────────────┐
        │          SatisfactoScript             │
        │  ├─ 1. Declarative Schema (dict)      │
        │  ├─ 2. Rule Registry (Python logic)   │
        │  ├─ 3. Delta I/O & Z-Order            │
        │  └─ 4. Semantic Layer + LLM Agent     │
        └──────────────────────────────────────┘
                          │
                 Power BI (Direct Query)

---

Installation

pip install satisfactoscript
pip install "satisfactoscript[spark]"        # Databricks + local PySpark/Delta

# Optional — LLM providers for the Semantic Layer
pip install satisfactoscript[llm-anthropic]   # Claude (Anthropic)
pip install satisfactoscript[llm-openai]      # GPT (OpenAI)
pip install satisfactoscript[llm-google]      # Gemini (Google)

# Optional — PDF export for session history
pip install satisfactoscript[semantic-pdf]

---

Local Development Setup

Run the same Spark-oriented framework on your laptop without a Databricks cluster. PySpark runs in local[*] mode, Delta Lake is enabled, and Apache Derby serves as the embedded metastore (no installation required).

1. Create config.yaml at your project root

default_env: LOCAL
priority_check: [DEV, QA, PROD, LOCAL]

environments:
  LOCAL:
    catalog: null          # null = no Unity Catalog, triggers local mode
    is_production: false

  DEV:
    catalog: "my_dev_catalog"
    is_production: false

  PROD:
    catalog: "my_prod_catalog"
    is_production: true

When catalog is null, the engine skips all Databricks catalog checks and boots in local mode. When Databricks credentials are present (DATABRICKS_HOST, DATABRICKS_TOKEN, DATABRICKS_CLUSTER_ID), the engine tries Databricks Connect first and falls back to local automatically.

2. (Optional) .env for Databricks credentials

# Only needed for Databricks Connect (remote cluster from IDE)
DATABRICKS_HOST=https://your-workspace.azuredatabricks.net
DATABRICKS_TOKEN=dapiXXXXXX
DATABRICKS_CLUSTER_ID=0123-456789-abcdef

# LLM provider for the Semantic Layer
ANTHROPIC_API_KEY=sk-ant-...

3. Boot the engine

from satisfactoscript import SatisfactoEngine

engine = SatisfactoEngine()  # auto-discovers config.yaml upwards from cwd
# → boots local[*] Spark + Delta Lake + Derby metastore
# → env=LOCAL, no catalog prefix, sandbox suffix = _<your_os_user>

Session detection priority:

1. Active Databricks session — running inside a notebook or cluster
2. Databricks Connect v2 — IDE + remote cluster via .env credentials
3. Local PySpark + Delta Lake — fully offline, zero configuration

---

Quick Start: Building a Pipeline

Option A — YAML file (recommended)

Store schemas as YAML files in your project for reuse and version control.

# schemas/gold/fact_transactions.yaml
tables:
  - name: "{{ catalog }}.silver.transactions"
    alias: tx
    filter:
      - "region:equals:EMEA"
      - "customer_id:is_not_null"
    quality_checks:
      drop_duplicates_on: [transaction_id]
      drop_nulls_in: [amount, transaction_date]

business_rules:
  - flag_high_value

select_final:
  - [transaction_id, id]
  - [transaction_date, date, [cast:date]]
  - [amount, amount_eur, [cast:double, round:2]]
  - [is_high_value, is_high_value]

from satisfactoscript import SatisfactoEngine, RuleRegistry, load_schema
from pyspark.sql import functions as F

engine = SatisfactoEngine()

@RuleRegistry.register_rule()
def flag_high_value(df):
    return df.withColumn("is_high_value", F.when(F.col("amount") >= 1000, 1).otherwise(0))

# Load schema from file — {{ catalog }} is replaced with engine's active catalog
schema = load_schema("schemas/gold/fact_transactions.yaml", params=engine.default_params)

# Preview before running (no Spark execution)
engine.describe_schema(schema)

# Run and write to Delta
engine.run_process_to_table(schema_dict=schema, target_layer="gold", target_table_name="fact_transactions")

Option B — Inline YAML string

For quick iterations or notebook-local schemas.

from satisfactoscript import SatisfactoEngine, parse_schema

engine = SatisfactoEngine()

schema = parse_schema("""
tables:
  - name: "{{ catalog }}.silver.transactions"
    alias: tx
    filter:
      - "region:equals:EMEA"
select_final:
  - [transaction_id, id]
  - [amount, amount_eur, [cast:double]]
""", params=engine.default_params)

engine.run_process_to_table(schema_dict=schema, target_layer="gold", target_table_name="fact_transactions")

Execution patterns

Method	Use case
run_process_to_table(schema, layer, table)	Process a schema and write to a single Delta table
run_process_and_split(schema, split_values, layer, base_name, col)	Split result into one table per value (e.g. one table per region)
run_union_sources_to_table(schema, partitions, src_layer, tgt_layer, table, bases, alias, dedup_after_union=True)	Union partitioned source tables, process, write
optimize_table(layer, table, zorder_cols)	Run Delta OPTIMIZE with optional ZORDER BY
describe_schema(schema)	Dry-run summary: sources, joins, columns — no Spark execution

---

Semantic Layer

The Semantic Layer lets you auto-generate structured YAML models from your Gold tables using an LLM, then query them in natural language.

Step 1 — Build a semantic model from a Gold table

SemanticBuilder inspects the table schema, optionally reads a Jupyter notebook and a business glossary, calls the LLM, validates the output (up to 3 attempts), and registers the model in semantic_catalog.yaml.

from satisfactoscript import SatisfactoEngine
from satisfactoscript.semantic.builder import SemanticBuilder
from satisfactoscript.semantic.llm_provider import get_llm_provider

engine = SatisfactoEngine()
builder = SemanticBuilder(
    llm_provider=get_llm_provider(),   # auto-detects from env vars
    output_dir="semantic_models",
)

builder.build(
    model_name="kpi_orders",
    split_value="erp",
    table="gold.fact_orders",
    layer="gold",
    source_notebook="notebooks/fact_orders.ipynb",  # optional — adds business context
    glossary_path="glossaries/orders.json",          # optional — injects domain terms
    description="Order KPIs from ERP (SAP source)",
    tags=["orders", "revenue", "erp"],
)
# → writes semantic_models/kpi_orders.erp.yaml
# → updates semantic_catalog.yaml

The generated YAML describes dimensions (with SQL expressions + types) and metrics (with SQL + aggregation type). It is fully human-readable and editable after generation.

Step 2 — Load the Semantic Engine

from satisfactoscript.semantic.semantic import SemanticEngine

sem = SemanticEngine(engine, models_dir="semantic_models")
# → loads semantic_catalog.yaml only at startup (lightweight)

# Browse available models
sem.list_models()
sem.list_models(tags=["revenue"], summary=True)

# Inspect a specific model
sem.get_model_summary("kpi_orders.erp")

Step 3 — Query in natural language with GenBIAgent

from satisfactoscript.agentic.agent import GenBIAgent

agent = GenBIAgent(semantic_engine=sem, llm_provider=get_llm_provider())

# Ask a business question
response = agent.ask("What is the total revenue by region for last quarter?")

if response.success:
    response.result.data.show()       # PySpark DataFrame
elif response.needs_clarification:
    print(response.clarification_message)

# Export the session as PDF
agent.history.to_pdf("session_export.pdf")

LLM provider auto-detection

get_llm_provider() selects the provider based on environment variables:

Variable	Provider
ANTHROPIC_API_KEY	Claude (Anthropic)
OPENAI_API_KEY	GPT (OpenAI)
GOOGLE_API_KEY	Gemini (Google)
LLM_PROVIDER=anthropic|openai|google	Force a specific provider

---

External Declarative Sources

Read external files (CSV, Parquet, JSON, Avro, ORC, Delta, Text) from any blob storage or local filesystem directly from your YAML schema — no Python required. All standard pipeline features (filter, quality checks, dev_limit, joins) apply identically to external sources.

tables:
  - name: raw_orders
    alias: orders
    source:
      type: csv
      path: "abfss://[e-mail]/bronze/orders/*.csv"
      options:
        header: "true"
        inferSchema: "true"
    filter:
      - "status:is_not_null"
    quality_checks:
      drop_nulls_in: [order_id, amount]
    dev_limit: 5000

  - name: "{{ catalog }}.silver.products"   # regular Delta table — mix freely
    alias: products

Supported type values: csv, parquet, json, avro, orc, delta, text.
Path supports {{ param }} injection — use load_schema(..., params={"base_path": "..."}) to parameterize.

Note: External sources are a Spark/Databricks feature. Legacy non-Spark backends are not part of the stabilized product path.

---

Adding Business Rules

Rules are decoupled from execution notebooks. Define them in a centralized rules.py and import it before running the engine.

from pyspark.sql import functions as F
from satisfactoscript import RuleRegistry

@RuleRegistry.register_rule()
def enrich_transaction_data(df):
    return (
        df
        .withColumn(
            "is_high_value",
            F.when(F.col("amount") >= 1000, 1).otherwise(0)
        )
        .withColumn(
            "clean_status",
            F.when(F.lower(F.col("status")).isin(["completed", "done"]), "Paid")
             .otherwise("Pending")
        )
        .fillna({"amount": 0.0})
    )

---

YAML Schema Reference

Filter operators

All operators use full English names. SQL abbreviations (eq, gte, etc.) are accepted as aliases.

Operator	Example	Notes
equals	"status:equals:ACTIVE"	alias: eq
not_equals	"status:not_equals:CANCELLED"	alias: ne
greater_than	"amount:greater_than:100"	alias: gt
less_than	"age:less_than:18"	alias: lt
greater_than_equal	"score:greater_than_equal:90"	alias: gte
less_than_equal	"qty:less_than_equal:5"	alias: lte
in	"status:in:ACTIVE,PENDING"	comma or ; separated
not_in	"region:not_in:FR,DE"
contains	"label:contains:promo"
not_contains	"label:not_contains:test"
starts_with	"ref:starts_with:ORD"
ends_with	"email:ends_with:@corp.com"
is_null	"discount:is_null"	no value
is_not_null	"customer_id:is_not_null"	no value
like	"name:like:J%"	SQL LIKE pattern
not_like	"name:not_like:test%"
sql	"sql:amount > threshold"	raw SQL escape hatch

For values containing commas, use the dict form: {column: city, operator: in, value: ["New York, NY", "Paris"]}.

select_final operations

Operations are applied left-to-right on the source column.

Operation	Example	Result
cast:type	cast:date, cast:double	Type casting
upper / lower	upper	String case
trim	trim	Strip whitespace
round:N	round:2	Round to N decimals
abs	abs	Absolute value
length	length	String length
to_date:fmt	to_date:yyyy-MM-dd	Parse string to date
nvl:val	nvl:0	Replace null with value
coalesce:val	coalesce:0	Same as nvl:
lit:val	lit:ERP	Constant value
expr:sql	expr:year(order_date)	Arbitrary SQL expression
split:sep,idx	split:-,1	Split string, take index
substring:start,len	substring:1,4	Substring
when:op:val	when:equals:DONE	Condition (use with then: / else:)

Shorthand for constant columns:

select_final:
  - [literal:ERP, source_system]          # adds source_system = 'ERP'
  - [literal:0.0, discount, [cast:double]]

OR filter groups:

filter_groups:
  - ["region:equals:EMEA", "status:is_not_null"]   # EMEA AND not null
  - ["region:equals:APAC"]                          # OR APAC

Keep all columns + add computed ones:

keep_all_columns: true
add_columns:
  - [amount, amount_rounded, [round:2]]
  - [literal:ERP, source_system]

Quality checks:

quality_checks:
  drop_nulls_in: [customer_id, order_date]
  drop_duplicates_on: [order_id, sku_id]

Compact join syntax:

join:
  - table_from: [orders, customer_id]
    table_to: [customers, id]
    type: left

Parameter injection:

tables:
  - name: "{{ catalog }}.silver.orders"
    filter:
      - "region:equals:{{ region }}"

schema = load_schema("schemas/fact_orders.yaml", params={**engine.default_params, "region": "FR"})

Dev sampling (ignored in job/prod):

dev_limit: 10000    # schema-level
tables:
  - name: silver.orders
    dev_limit: 5000  # table-level override

---

Smart Sandbox

In interactive (non-job, non-prod) mode, source tables are transparently resolved to your personal sandbox schema (schema_XXXX where XXXX is derived from your username).

Situation	Behavior
Table exists in silver_XXXX	Loaded directly — transparent
Table missing in silver_XXXX	Logged warning + shallow clone from silver + load
Schema silver_XXXX doesn't exist	Schema created + table cloned + load
Table missing in main schema	ValueError raised

Configure behavior in config.yaml:

sandbox:
  missing_table: copy   # copy (default) | error

Note: Add .satisfacto_user to your .gitignore. This file is auto-generated to cache your sandbox suffix.

---

Developer Tools

# Force a specific environment (bypass auto-detection)
engine = SatisfactoEngine(force_env="LOCAL")

# Preview a schema without executing Spark
engine.describe_schema(schema)

# List registered rules and loaders
RuleRegistry.list_rules()
RuleRegistry.list_loaders()

---

Configuration Reference (config.yaml)

default_env: LOCAL                      # Fallback if no Databricks catalog is reachable
priority_check: [DEV, QA, PROD, LOCAL]  # Detection order

environments:
  LOCAL:
    catalog: null                       # null = local mode (no Unity Catalog)
    is_production: false

  DEV:
    catalog: "my_dev_catalog"
    is_production: false

  QA:
    catalog: "my_qa_catalog"
    is_production: false

  PROD:
    catalog: "my_prod_catalog"
    is_production: true

sandbox:
  missing_table: copy                   # copy (default) | error

The engine also reads the semantic_views_schema key (optional) to resolve the target schema for semantic views.