Usage guide¶

1. Initialise the platform¶

from scamander.processor.platform import Platform

# Auto-detects Fabric, Databricks, or vanilla Spark.
# Sets up both human-readable and machine-readable logging automatically —
# console log calls are forwarded into the machine log table too.
platform = Platform(
    logging_catalog="main",             # Unity Catalog / catalog name (default: "main")
    logging_schema="mdif",              # Schema for the logging table (default: "mdif")
    logging_table="logging",            # Delta table name (default: "logging")
    retention_days=90,                  # Machine log rows older than this are deleted on init (default: 90)
    log_level="INFO",                   # Minimum level written to the machine log table (default: "INFO")
    log_timezone="Europe/Amsterdam",    # Timezone for machine log timestamps (default: "Europe/Amsterdam")
)

2. Read a config file¶

# JSON — returns a dict or list
config_data = platform.read_config(
    "abfss://container@account.dfs.core.windows.net/configs/ingest.json"
)

# CSV — returns a list of dicts, one per row
config_data = platform.read_config("dbfs:/mnt/configs/mappings.csv")

# YAML — returns a dict or list
config_data = platform.read_config(
    "abfss://container@account.dfs.core.windows.net/configs/pipeline.yaml"
)

read_config validates that the file exists before reading and raises FileNotFoundError if it is missing.

3. Read data from a source¶

Using the Source factory (recommended): Source(config) inspects config.source.type and returns the correct connector automatically, so your code never needs to know which connector is in use.

from scamander.models import Config
from scamander.models.source_config import SourceConfig
from scamander.models.source_config.csv import CsvSourceOptions
from scamander.models.sink_config import SinkConfig
from scamander.models.sink_config.lakehouse import LakeHouseSinkOptions
from scamander.sources import Source

config = Config(
    source=SourceConfig(type="csv", options=CsvSourceOptions()),
    sink=SinkConfig(type="LakeHouse", options=LakeHouseSinkOptions()),
)

connector = Source(config)   # returns a CsvConnector
df = connector.connect()     # returns a Spark DataFrame

Instantiating a connector directly:

from scamander.sources.csv_connector import CsvConnector

connector = CsvConnector(
    file_path="abfss://raw@account.dfs.core.windows.net/exports/sales_20250315.csv",
    header=True,
    separator=",",
)
df = connector.connect()

Oracle and PostgreSQL sources connect over JDBC. For Oracle, both service_name and sid are supported (service_name takes precedence when both are set), and a query is automatically wrapped in a subquery alias for JDBC compatibility:

from scamander.sources.oracle_connector import OracleConnector
from scamander.models.source_config.oracle import OracleSourceOptions

oracle_options = OracleSourceOptions(
    host="oracle.example.com",
    port=1521,
    service_name="ORCL",   # or sid="ORCL" for legacy SID-style connections
    username="hr",
    password="s3cret",
    table="HR.EMPLOYEES",  # or query="SELECT * FROM HR.EMPLOYEES WHERE active=1"
    fetch_size=10000,       # rows per JDBC round-trip (default)
)

4. Write data to a Delta table¶

# mode defaults to "overwrite"; schema is created automatically if missing
success = platform.write_to_table(df, full_table_name="bronze.customer", mode="append")

# Supported modes: "overwrite" (default), "append", "error", "ignore"
platform.write_to_table(df, "silver.orders", mode="overwrite")

5. Create a table from metadata¶

metadata = platform.read_config("/configs/customer_table.json")

# Generates and executes: CREATE TABLE IF NOT EXISTS ... USING DELTA
platform.create_empty_table_from_metadata(metadata)

With SCD Type-2 history tracking enabled in the config ("track_history": true), the five SCD Type-2 columns HashSleutel, HashWaarde, GeldigVan, GeldigTot, and IsActief are appended to the generated DDL automatically — see API reference for the full column definitions.

6. Bronze → Silver transformations¶

Use Transformer to clean and standardise raw bronze data before writing to silver. All steps preserve Spark’s lazy evaluation: no actions trigger until .df is consumed by a write.

from scamander.transformations import Transformer

bronze_df = platform.spark.read.table("bronze.customer")
metadata  = platform.read_config("/configs/customer_table.json")

silver_df = (
    Transformer(bronze_df, log=platform.log)
    .cast_to_schema(metadata["columns"])                              # cast to metadata types
    .rename_columns({"klant_id": "customer_id", "klant_naam": "customer_name"})
    .trim_strings()
    .replace_empty_strings()
    .filter_nulls(["customer_id"])                                    # drop rows missing the PK
    .drop_duplicates(subset=["customer_id"])                          # dedup on the natural key
    .add_audit_columns(source_file="customer_export.csv")             # lineage columns
    .df
)

platform.write_to_table(silver_df, "silver.customer", mode="overwrite")

Available cleaning methods¶
Method	Description
`cast_to_schema(metadata)`	Cast columns to types declared in a metadata column list
`rename_columns(mapping)`	Rename via a `{source: target}` dict
`select_columns(columns)`	Drop all columns not in the list
`trim_strings()`	Strip leading/trailing whitespace from all STRING columns
`replace_empty_strings(replacement)`	Convert `""` to null (or a custom value) in all STRING columns
`filter_nulls(columns)`	Remove rows where any listed column is null
`drop_duplicates(subset)`	Deduplicate; scope to a subset of columns when provided
`standardize_column_names()`	Lowercase and snake_case all column names
`add_audit_columns(source_file, batch_id)`	Append `_ingestion_timestamp`, `_source_file`, `_batch_id`

7. Silver → Gold transformations¶

orders_df    = platform.spark.read.table("silver.orders")
customers_df = platform.spark.read.table("silver.customer")

gold_df = (
    Transformer(orders_df, log=platform.log)
    .join(customers_df, on="customer_id", how="left", broadcast_right=True)
    .add_derived_columns({"revenue_incl_vat": "revenue * 1.21"})
    .aggregate(
        group_by=["region", "product_category"],
        aggregations=[
            {"column": "revenue",  "function": "sum",           "alias": "total_revenue"},
            {"column": "order_id", "function": "count_distinct", "alias": "unique_orders"},
            {"column": "revenue",  "function": "avg",           "alias": "avg_order_value"},
        ],
    )
    .df
)

platform.write_to_table(gold_df, "gold.revenue_by_region", mode="overwrite")

Available enrichment methods¶
Method	Description
`join(other, on, how, broadcast_right)`	Join with another DataFrame; broadcast hint for small lookup tables
`aggregate(group_by, aggregations)`	GroupBy + multi-function aggregation from a spec list
`add_derived_columns(expressions)`	Add columns from Column objects or SQL expression strings
`deduplicate_with_ranking(partition_by, order_by, descending)`	Keep rank-1 row per partition via `row_number()`
`pivot_table(group_by, pivot_col, agg_col, agg_func, pivot_values)`	Pivot a column into multiple columns

Supported aggregation functions: sum, count, avg, mean, min, max, first, last, count_distinct, stddev, variance, collect_list, collect_set.

8. SCD Type-2 merges¶

Use apply_scd2_merge when the target table has track_history: true and you want to version entity changes instead of overwriting:

from scamander.transformations.scd import apply_scd2_merge

# source_df contains the latest state of each entity
counts = apply_scd2_merge(
    spark=platform.spark,
    source_df=silver_df,
    target_table="silver.customer",
    key_columns=["customer_id"],                              # uniquely identifies an entity
    tracked_columns=["customer_name", "region", "revenue"],   # changes here create new versions
)

print(f"Expired: {counts['closed']}  Inserted: {counts['inserted']}")

The merge runs in two steps:

Expire active rows whose tracked-column hash has changed (IsActief = false, GeldigTot = now).
Insert new rows for all source records that have no currently active version (covers new entities and just-expired records).

compute_hash is also available standalone when you need SHA-256 hashes in other contexts:

from scamander.transformations.scd import compute_hash

df = compute_hash(df, columns=["customer_id"], alias="HashSleutel")
df = compute_hash(df, columns=["name", "region"], alias="HashWaarde")

9. Config-driven transformation pipelines¶

Every transformation step can be described in a JSON/YAML config file and parsed via TransformationConfig:

{
  "source_table": "bronze.customer",
  "target_table": "silver.customer",
  "column_mappings": [
    { "source": "klant_id",   "target": "customer_id",   "type": "INT",     "nullable": false },
    { "source": "klant_naam", "target": "customer_name", "type": "VARCHAR"                    }
  ],
  "trim_strings": true,
  "replace_empty_strings": true,
  "filter_null_columns": ["customer_id"],
  "drop_duplicates": true,
  "duplicate_subset": ["customer_id"],
  "add_audit_columns": true,
  "write_mode": "overwrite"
}

from scamander.models import TransformationConfig

raw = platform.read_config("/configs/bronze_to_silver_customer.json")
cfg = TransformationConfig(**raw)

For silver → gold with joins and aggregations, the config additionally supports joins, group_by, and aggregations keys.

10. File utilities¶

# Check if a file exists — Returns True / False, or raises immediately with raise_exceptions=True
exists = platform.check_file_exists("abfss://container@account.dfs.core.windows.net/data/file.csv")
platform.check_file_exists("dbfs:/mnt/raw/file.csv", raise_exceptions=True)

# Resolve a wildcard file mask — returns a sorted list of matching filenames
files = platform.resolve_filemask(
    "abfss://raw@account.dfs.core.windows.net/exports/sales_*.csv"
)

# Pick the most-recent file using an embedded timestamp
# (filenames must contain a 12-digit YYYYMMDDHHMI timestamp)
files.sort(key=Platform.extract_datetime_from_filename)
latest_file = files[-1]

# Move a file
platform.move_file(
    source="abfss://raw@account.dfs.core.windows.net/inbox/data.csv",
    destination="abfss://raw@account.dfs.core.windows.net/archive/data.csv",
    overwrite=True,
    create_missing_dirs=True,
)

Extract an archive: supported extensions are .zip, .tar, .tar.gz/.tgz, and .tar.bz2/.tbz2. Every archive member path is resolved before extraction; members that would write outside the destination directory are rejected with a ValueError (zip-slip protection).

from scamander.utils import extract_archive

extracted_files = extract_archive(
    archive_path="/lakehouse/default/Files/Inbox/export.tar.gz",
    destination="/lakehouse/default/Files/Inbox/export/",
    log=platform.log,   # optional — logs file count on completion
)

On Fabric, combine with Fabric.to_local_path to go from an ABFS path to a local path first:

from scamander.processor.fabric import Fabric
from scamander.utils import extract_archive, read_json_records

local_archive = Fabric.to_local_path(f"{ABFS_PREFIX}/Files/Inbox/brp_bulk.zip")
local_dir     = local_archive.replace(".zip", "/")

files   = extract_archive(local_archive, local_dir, log=platform.log)
records = read_json_records(local_dir, log=platform.log)

11. Logging¶

The framework sets up both loggers automatically when Platform() is initialised, and attaches a MachineLogHandler to Platform.log so every console log call is also persisted to the machine log table (subject to log_level filtering). Either logger can also be used independently.

HumanLogger — structured stdout output, suitable for notebook cells:

from scamander.logging.data_pipeline_logger import HumanLogger

log = HumanLogger(name="my_pipeline")
log.info("Starting ingestion run")
log.warning("Column 'age' contains nulls")
log.error("Schema mismatch detected")

MachineLogger — persists structured records to a Delta table for operational observability:

from scamander.logging.machine_logger import MachineLogger

mlog = MachineLogger(
    notebook_name="Ingest_Customer",
    notebook_path="https://app.fabric.microsoft.com/...",
    session_id="sess-abc-123",
    logging_catalog="main",
    logging_schema="mdif",
    logging_table="logging",
    log_level="INFO",                 # minimum level written to the table (default: "INFO")
    timezone="Europe/Amsterdam",      # timezone for the timestamp column (default: "Europe/Amsterdam")
)

mlog.info(message="Loaded 50,000 rows", object_name="bronze.customer", total_records=50000, status="Success")
mlog.error(message="File not found", metadata_config_file="/configs/ingest.json", status="Failed")

Records below log_level are dropped before any Spark work happens. This applies both to direct MachineLogger calls and to records forwarded from HumanLogger, independently of HumanLogger’s own stdlib logging level, so Platform(log_level="ERROR") still prints INFO messages; they just won’t reach the table.

MachineLogger.cleanup_old_records(retention_days) deletes table rows older than retention_days. Platform() calls this once per initialisation using its own retention_days argument (default 90); failures are logged as a warning rather than aborting startup.

12. OAuth 2.0 API calls¶

OAuth2Session is a drop-in replacement for requests.Session when the target API requires a client-credentials token. Tokens are fetched on construction and refreshed automatically:

from scamander.api.oauth2_session import OAuth2Session

session = OAuth2Session(
    token_url="https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token",
    client_id="my-app-client-id",
    client_secret="my-app-client-secret",
    token_refresh_margin=60,  # proactively refresh 60 seconds before expiry
)

# Use exactly like requests.Session — the Bearer token is injected automatically
response = session.get("https://api.example.com/v1/entities")
response = session.post("https://api.example.com/v1/entities", json={"key": "value"})

13. Data quality checks¶

Scamander integrates Great Expectations out of the box, via an ephemeral (in-memory) GE context. No filesystem store or GE project setup is required, which keeps it suitable for Fabric and Databricks notebooks.

Quality suites are defined as QualitySuiteConfig objects (or loaded from JSON/YAML):

from scamander.models.quality_config import ExpectationSpec, QualitySuiteConfig

suite = QualitySuiteConfig(
    name="customer_bronze_quality",
    raise_on_failure=False,   # True -> raises DataQualityError on failure
    expectations=[
        ExpectationSpec(type="not_null", column="customer_id"),
        ExpectationSpec(type="unique",   column="customer_id"),
        ExpectationSpec(type="in_set",   column="status", value_set=["active", "inactive"]),
        ExpectationSpec(type="regex",    column="email",  regex=r"^[^@]+@[^@]+\.[^@]+$"),
        ExpectationSpec(type="between",  column="age",    min_value=0, max_value=120),
        ExpectationSpec(type="row_count_between", min_value=1),
        # Allow up to 2% nulls in a non-critical column
        ExpectationSpec(type="not_null", column="phone", mostly=0.98),
    ],
)

Run it via platform.validate_dataframe(df, suite), as a terminal step in a Transformer chain (.validate(suite) — returns a QualityResult, not self), or directly via QualityChecker(spark, log).check(df, suite).

Supported expectation types¶
`type`	Required fields	Description
`not_null`	`column`	Column must contain no nulls (honoured per row via `mostly`)
`unique`	`column`	All values in the column must be distinct
`multi_column_unique`	`columns`	Combination of columns must be unique per row
`between`	`column`, `min_value`, `max_value`	Values must fall within the given bounds
`in_set` / `not_in_set`	`column`, `value_set`	Values must (not) be members of the given set
`regex`	`column`, `regex`	Values must match the regular expression
`row_count_between`	`min_value` and/or `max_value`	Table row count must be within bounds
`column_exists`	`column`	Column must be present in the DataFrame schema
`column_type`	`column`, `type_`	Column must have the specified Spark type (e.g. `"StringType"`)
`foreign_key`	`column`, `ref_table`, `ref_column`	Every non-null value in `column` must exist in `ref_column` of `ref_table` (a registered Spark table or temp view)
`column_comparison`	`expression`	A SQL boolean expression must hold for every row (e.g. `"start_date <= end_date"`)
`conditional_not_null`	`when_column`, `then_column`	When `when_column` is not null, `then_column` must also be not null
`custom_sql`	`sql`	Spark-SQL query that must return zero rows to pass; use `{batch}` to reference the DataFrame being validated

All column-level expectations accept an optional mostly field (float 0.0–1.0, default 1.0) allowing a configurable fraction of rows to fail before the expectation is marked failed. The four SQL-backed types (foreign_key, column_comparison, conditional_not_null, custom_sql) are all-or-nothing and don’t support mostly.

Tracking history with QualityReporter: persists quality run results to Delta tables so you can track compliance over time and drill into the exact rows that violated each check.

from scamander.quality import QualityChecker, QualityReporter

checker  = QualityChecker(spark=platform.spark, log=platform.log)
reporter = QualityReporter(
    spark=platform.spark,
    history_table="dq.quality_history",       # required — one row per expectation per run
    violations_table="dq.quality_violations",  # optional — one row per violating row
    log=platform.log,
)

result = checker.check(df, suite)
run_id = reporter.save(df, result, suite, dataset_name="customers_bronze")

Violations are written for every expectation with at least one violating row, regardless of whether it passed via mostly. Schema-level checks (row_count_between, column_exists, column_type) never produce violation rows. See API reference for the history/violations table schemas.

14. Anonymization / pseudonymization¶

Scamander provides a declarative anonymization layer for protecting PII before writing data to downstream tables. hash, mask, redact, and generalize use only PySpark built-ins; encrypt and fake require pip install "Scamander-MDIF[anonymization]".

Method	Technique	Reversible
`hash`	SHA-256 or HMAC-SHA256	No
`mask`	Character replacement with optional prefix/suffix	No
`redact`	Replace with `null`	No
`generalize`	Bucket numeric values into labelled ranges	No
`encrypt`	AES-256-GCM authenticated encryption	Yes (with key)
`fake`	Faker-generated synthetic data	No

from scamander.models.anonymization_config import AnonymizationConfig, AnonymizationRule

config = AnonymizationConfig(
    name="customer_pii",
    rules=[
        AnonymizationRule(column="email", method="hash"),                                      # SHA-256
        AnonymizationRule(column="bsn", method="hash", hash_key="my-secret"),                    # keyed HMAC
        AnonymizationRule(column="card_number", method="mask", visible_suffix=4),                # keep last 4
        AnonymizationRule(column="phone", method="mask", visible_prefix=3, mask_char="X"),
        AnonymizationRule(column="birth_date", method="redact"),
        AnonymizationRule(
            column="age", method="generalize",
            bins=[0.0, 18.0, 65.0, 120.0], labels=["child", "adult", "senior"],
        ),
        AnonymizationRule(column="iban", method="encrypt", encrypt_key="<base64-32-byte-key>"),  # requires [anonymization]
        AnonymizationRule(column="full_name", method="fake", faker_provider="name"),             # requires [anonymization]
    ],
)

Generate a one-off AES-256 key with base64.b64encode(os.urandom(32)).

Use Anonymizer(log=platform.log).apply(df, config) directly, or the .anonymize(config) step inside a Transformer chain (returns self, so further steps can follow):

silver_df = (
    Transformer(bronze_df, log=platform.log)
    .cast_to_schema(metadata["columns"])
    .trim_strings()
    .filter_nulls(["customer_id"])
    .drop_duplicates(subset=["customer_id"])
    .anonymize(config)          # apply all PII rules in one step
    .add_audit_columns(source_file="customers.csv")
    .df
)

15. Secret providers¶

Pass a SecretProvider subclass to Platform() to enable platform.get_secret(). Without one, get_secret raises NotImplementedError.

from scamander.secrets import AzureKeyVaultProvider

platform = Platform(
    secret_provider=AzureKeyVaultProvider(default_vault="my-keyvault"),
    ...
)
password = platform.get_secret("pg-password")

# Supply a custom credential (e.g. from Databricks Service Credentials):
from databricks.sdk.credentials_provider import databricks_sdk_credentials
credential = databricks_sdk_credentials(...)
platform = Platform(
    secret_provider=AzureKeyVaultProvider(default_vault="my-keyvault", credential=credential),
    ...
)

Install the required extra first, e.g. pip install 'Scamander-MDIF[azure-keyvault]'. See API reference for the full list of providers (HashiCorp Vault, Azure Key Vault, Databricks secret scopes) and their constructor options.