Gebruikshandleiding¶

1. Het platform initialiseren¶

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. Een configbestand lezen¶

# 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 valideert dat het bestand bestaat voordat het wordt gelezen en werpt FileNotFoundError op als het ontbreekt.

3. Data lezen uit een bron¶

Met de Source``**-factory (aanbevolen)**: ``Source(config) inspecteert config.source.type en geeft automatisch de juiste connector terug, zodat je code nooit hoeft te weten welke connector wordt gebruikt.

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

Een connector direct instantiëren:

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- en PostgreSQL-bronnen verbinden over JDBC. Voor Oracle worden zowel service_name als sid ondersteund (service_name heeft voorrang als beide zijn ingesteld), en een query wordt automatisch in een subquery-alias gewikkeld voor JDBC-compatibiliteit:

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. Data wegschrijven naar een Delta-tabel¶

# 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. Een tabel aanmaken vanuit 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)

Met SCD Type-2 historietracking ingeschakeld in de config ("track_history": true), worden de vijf SCD Type-2-kolommen HashSleutel, HashWaarde, GeldigVan, GeldigTot, en IsActief automatisch toegevoegd aan de gegenereerde DDL — zie API-referentie voor de volledige kolomdefinities.

6. Bronze → Silver-transformaties¶

Gebruik Transformer om ruwe bronze-data te reinigen en standaardiseren voordat je naar silver schrijft. Alle stappen behouden Sparks lazy evaluation: er worden geen acties getriggerd totdat .df door een write wordt geconsumeerd.

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")

Beschikbare cleaning-methodes¶
Methode	Beschrijving
`cast_to_schema(metadata)`	Cast kolommen naar de types opgegeven in een metadata-kolomlijst
`rename_columns(mapping)`	Hernoem via een `{source: target}`-dict
`select_columns(columns)`	Verwijder alle kolommen die niet in de lijst staan
`trim_strings()`	Verwijder leidende/afsluitende witruimte uit alle STRING-kolommen
`replace_empty_strings(replacement)`	Zet `""` om naar null (of een custom waarde) in alle STRING-kolommen
`filter_nulls(columns)`	Verwijder rijen waarin een van de opgegeven kolommen null is
`drop_duplicates(subset)`	Dedupliceer; beperk tot een subset van kolommen indien opgegeven
`standardize_column_names()`	Zet alle kolomnamen om naar kleine letters en snake_case
`add_audit_columns(source_file, batch_id)`	Voeg `_ingestion_timestamp`, `_source_file`, `_batch_id` toe

7. Silver → Gold-transformaties¶

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")

Beschikbare enrichment-methodes¶
Methode	Beschrijving
`join(other, on, how, broadcast_right)`	Join met een andere DataFrame; broadcast-hint voor kleine lookup-tabellen
`aggregate(group_by, aggregations)`	GroupBy + multi-functie-aggregatie vanuit een spec-lijst
`add_derived_columns(expressions)`	Voeg kolommen toe vanuit Column-objecten of SQL-expressiestrings
`deduplicate_with_ranking(partition_by, order_by, descending)`	Houd de rank-1-rij per partitie via `row_number()`
`pivot_table(group_by, pivot_col, agg_col, agg_func, pivot_values)`	Pivoteer een kolom naar meerdere kolommen

Ondersteunde aggregatiefuncties: sum, count, avg, mean, min, max, first, last, count_distinct, stddev, variance, collect_list, collect_set.

8. SCD Type-2-merges¶

Gebruik apply_scd2_merge wanneer de doeltabel track_history: true heeft en je entiteitswijzigingen wilt versioneren in plaats van overschrijven:

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']}")

De merge verloopt in twee stappen:

Verloop actieve rijen waarvan de hash van de gevolgde kolom is veranderd (IsActief = false, GeldigTot = now).
Voeg nieuwe rijen toe voor alle bronrecords zonder momenteel actieve versie (dekt nieuwe entiteiten en net verlopen records).

compute_hash is ook standalone beschikbaar wanneer je in andere contexten SHA-256-hashes nodig hebt:

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-gestuurde transformatiepipelines¶

Elke transformatiestap kan worden beschreven in een JSON-/YAML-configbestand en geparst 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)

Voor silver → gold met joins en aggregaties ondersteunt de config bovendien de keys joins, group_by, en aggregations.

10. Bestandshulpprogramma’s¶

# 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,
)

Een archief uitpakken: ondersteunde extensies zijn .zip, .tar, .tar.gz/.tgz en .tar.bz2/.tbz2. Het pad van elk archiefonderdeel wordt vóór het uitpakken opgelost; onderdelen die buiten de doelmap zouden schrijven worden geweigerd met een ValueError (zip-slip-bescherming).

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
)

Combineer op Fabric met Fabric.to_local_path om eerst van een ABFS-pad naar een lokaal pad te gaan:

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¶

Het framework zet beide loggers automatisch op wanneer Platform() wordt geïnitialiseerd, en koppelt een MachineLogHandler aan Platform.log zodat elke console-logaanroep ook wordt weggeschreven naar de machine-logtabel (afhankelijk van log_level-filtering). Beide loggers kunnen ook onafhankelijk worden gebruikt.

HumanLogger — gestructureerde stdout-output, geschikt voor notebookcellen:

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 — schrijft gestructureerde records weg naar een Delta-tabel voor operationele 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 onder log_level worden verwijderd voordat er Spark-werk plaatsvindt. Dit geldt zowel voor directe MachineLogger-aanroepen als voor records die worden doorgestuurd vanuit HumanLogger, onafhankelijk van het eigen stdlib-logniveau van HumanLogger. Platform(log_level="ERROR") print dus nog steeds INFO-berichten; ze bereiken alleen de tabel niet.

MachineLogger.cleanup_old_records(retention_days) verwijdert tabelrijen ouder dan retention_days. Platform() roept dit één keer per initialisatie aan met zijn eigen retention_days-argument (standaard 90); fouten worden gelogd als waarschuwing in plaats van dat de opstart wordt afgebroken.

12. OAuth 2.0 API-calls¶

OAuth2Session is een drop-in vervanging voor requests.Session wanneer de doel-API een client-credentials-token vereist. Tokens worden bij constructie opgehaald en automatisch ververst:

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. Datakwaliteitscontroles¶

Scamander integreert Great Expectations standaard, via een ephemere (in-memory) GE-context. Er is geen filesystem store of GE-projectconfiguratie nodig, wat het geschikt houdt voor Fabric- en Databricks-notebooks.

Quality-suites worden gedefinieerd als QualitySuiteConfig-objecten (of geladen vanuit 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),
    ],
)

Voer het uit via platform.validate_dataframe(df, suite), als afsluitende stap in een Transformer-chain (.validate(suite) — retourneert een QualityResult, geen self), of direct via QualityChecker(spark, log).check(df, suite).

Ondersteunde expectation-types¶
`type`	Verplichte velden	Beschrijving
`not_null`	`column`	Kolom mag geen nulls bevatten (per rij gehonoreerd via `mostly`)
`unique`	`column`	Alle waarden in de kolom moeten uniek zijn
`multi_column_unique`	`columns`	Combinatie van kolommen moet uniek zijn per rij
`between`	`column`, `min_value`, `max_value`	Waarden moeten binnen de opgegeven grenzen vallen
`in_set` / `not_in_set`	`column`, `value_set`	Waarden moeten (niet) lid zijn van de opgegeven set
`regex`	`column`, `regex`	Waarden moeten overeenkomen met de reguliere expressie
`row_count_between`	`min_value` en/of `max_value`	Rijaantal van de tabel moet binnen de grenzen vallen
`column_exists`	`column`	Kolom moet aanwezig zijn in het DataFrame-schema
`column_type`	`column`, `type_`	Kolom moet het opgegeven Spark-type hebben (bijv. `"StringType"`)
`foreign_key`	`column`, `ref_table`, `ref_column`	Elke niet-null waarde in `column` moet bestaan in `ref_column` van `ref_table` (een geregistreerde Spark-tabel of temp view)
`column_comparison`	`expression`	Een SQL boolean-expressie moet gelden voor elke rij (bijv. `"start_date <= end_date"`)
`conditional_not_null`	`when_column`, `then_column`	Wanneer `when_column` niet null is, mag `then_column` ook niet null zijn
`custom_sql`	`sql`	Spark-SQL-query die nul rijen moet retourneren om te slagen; gebruik `{batch}` om te verwijzen naar de DataFrame die wordt gevalideerd

Alle expectations op kolomniveau accepteren een optioneel veld mostly (float 0.0–1.0, standaard 1.0) waarmee een configureerbare fractie rijen mag falen voordat de expectation als gefaald wordt gemarkeerd. De vier SQL-gebaseerde types (foreign_key, column_comparison, conditional_not_null, custom_sql) zijn alles-of-niets en ondersteunen mostly niet.

Historie bijhouden met QualityReporter: schrijft resultaten van quality-runs weg naar Delta-tabellen zodat je compliance in de tijd kunt volgen en kunt inzoomen op de exacte rijen die elke check overtraden.

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 worden weggeschreven voor elke expectation met minstens één overtredende rij, ongeacht of deze slaagde via mostly. Checks op schemaniveau (row_count_between, column_exists, column_type) produceren nooit violation-rijen. Zie API-referentie voor de tabelschema’s van history/violations.

14. Anonimisering / pseudonimisering¶

Scamander biedt een declaratieve anonimiseringslaag voor het beschermen van PII voordat data naar downstream-tabellen wordt weggeschreven. hash, mask, redact, en generalize gebruiken alleen PySpark-built-ins; encrypt en fake vereisen pip install "Scamander-MDIF[anonymization]".

Methode	Techniek	Omkeerbaar
`hash`	SHA-256 of HMAC-SHA256	Nee
`mask`	Tekenvervanging met optionele prefix/suffix	Nee
`redact`	Vervang door `null`	Nee
`generalize`	Bucket numerieke waarden in gelabelde bereiken	Nee
`encrypt`	AES-256-GCM authenticated encryption	Ja (met sleutel)
`fake`	Door Faker gegenereerde synthetische data	Nee

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]
    ],
)

Genereer een eenmalige AES-256-sleutel met base64.b64encode(os.urandom(32)).

Gebruik Anonymizer(log=platform.log).apply(df, config) direct, of de stap .anonymize(config) binnen een Transformer-chain (retourneert self, zodat verdere stappen kunnen volgen):

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¶

Geef een SecretProvider-subklasse door aan Platform() om platform.get_secret() te activeren. Zonder een subklasse werpt get_secret NotImplementedError op.

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),
    ...
)

Installeer eerst de vereiste extra, bijv. pip install 'Scamander-MDIF[azure-keyvault]'. Zie API-referentie voor de volledige lijst van providers (HashiCorp Vault, Azure Key Vault, Databricks secret scopes) en hun constructor-opties.