15. Trust Through Traceability: Advanced Validation and OpenLineage Integration

---

🔍 Why Validation Alone Is Not Enough

If you are building a regulated financial data pipeline, simple schema validation is not enough — it misses semantic errors, statistical anomalies, and the lineage evidence that GDPR auditors and compliance teams require. This section explains why the platform uses three distinct validation layers and what each one catches that the others cannot.

Most platforms validate schema: is the amount field a number? Is currency a three-character string? This catches format errors but misses semantic errors: an amount of -99999 is a valid number but an invalid transaction. A currency of "ZZZ" is a valid three-character string but not a recognised currency code.

Statistical validation goes further: is this batch’s amount distribution consistent with the last 30 days? If the mean amount has jumped by 400%, that might be legitimate (a large corporate transaction day) or it might indicate a data source problem (a currency conversion stage applied the wrong rate).

And none of these tells you where the data came from, what transformations it passed through, or why a specific record has the values it has. For GDPR compliance and financial audit, the ability to explain how a piece of data arrived at its current state is not optional.

The platform addresses all three requirements — schema validation, semantic validation, statistical validation, and provenance — in a layered validation architecture that reports through a shared alerting callback.

↑ Back to top · Next: The Validation Stage →

---

⚙️ The Validation Stage

The ValidationStage applies three validation layers in sequence:

class ValidationStage(PipelineStage):
    def __init__(self, config: dict | None = None):
        self.config = config or {}
        self.schema_validator = SchemaValidator()
        self.rules_validator = BusinessRulesValidator(
            rules=self.config.get("business_rules", DEFAULT_BUSINESS_RULES)
        )
        self.statistical_validator = StatisticalValidator(
            lookback_days=self.config.get("lookback_days", 30)
        )
        self.alert_callback = self.config.get("alert_callback")

    def name(self) -> str:
        return "ValidationStage"

    def execute(
        self,
        data: tuple[list[Transaction], list[str]],
        pipeline_context: dict,
    ) -> tuple[list[Transaction], list[str]]:
        transactions, memos = data

        schema_result = self.schema_validator.validate_batch(transactions)
        rules_result = self.rules_validator.validate_batch(transactions)
        statistical_result = self.statistical_validator.validate_batch(
            transactions,
            batch_id=pipeline_context.get("batch_id"),
        )

        all_issues = (
            schema_result.issues
            + rules_result.issues
            + statistical_result.issues
        )

        if all_issues:
            self._record_metrics(all_issues)
            if self.alert_callback:
                self.alert_callback(
                    severity="warning" if not any(i.blocking for i in all_issues) else "error",
                    message=f"Validation found {len(all_issues)} issues in batch {pipeline_context.get('batch_id')}",
                    source="ValidationStage",
                    details={"issues": [i.to_dict() for i in all_issues]},
                )

        blocking_issues = [i for i in all_issues if i.blocking]
        if blocking_issues and self.config.get("fail_on_blocking", True):
            raise ValidationError(
                f"Batch {pipeline_context.get('batch_id')} has {len(blocking_issues)} blocking validation failures"
            )

        pipeline_context["validation_result"] = {
            "passed": len(blocking_issues) == 0,
            "total_issues": len(all_issues),
            "blocking_issues": len(blocking_issues),
            "schema_issues": len(schema_result.issues),
            "rules_issues": len(rules_result.issues),
            "statistical_issues": len(statistical_result.issues),
        }

        return transactions, memos

The key design decision: validation issues are separated into blocking and non-blocking. A record with an invalid currency code is blocking — it must not proceed through the pipeline. A statistical anomaly in the batch’s amount distribution is non-blocking — it should be flagged and investigated, but the records are structurally valid and should be ingested. Treating all issues as blocking creates alert fatigue; treating all issues as warnings creates silent data quality degradation.

↑ Back to top · Next: Schema Validation →

---

🧩 Schema Validation

Schema validation uses Pydantic‘s model validation rather than a separate schema checker. Every Transaction is already a Pydantic model; schema validation means checking constraints that Pydantic enforces at instantiation, plus additional checks for values that pass Pydantic’s type constraints but fail domain constraints:

class SchemaValidator:
    VALID_CURRENCIES = set(get_supported_currencies())

    def validate_batch(self, transactions: list[Transaction]) -> ValidationResult:
        issues = []
        for i, tx in enumerate(transactions):
            if tx.currency not in self.VALID_CURRENCIES:
                issues.append(ValidationIssue(
                    record_index=i,
                    field="currency",
                    value=tx.currency,
                    rule="valid_currency",
                    message=f"Currency '{tx.currency}' is not a recognised ISO 4217 code",
                    blocking=True,
                ))
            if tx.transaction_type not in TRANSACTION_TYPES:
                issues.append(ValidationIssue(
                    record_index=i,
                    field="transaction_type",
                    value=tx.transaction_type,
                    rule="valid_transaction_type",
                    message=f"Transaction type '{tx.transaction_type}' is not registered",
                    blocking=True,
                ))
        return ValidationResult(issues=issues)

The VALID_CURRENCIES set is loaded at startup from the platform’s configuration, not hardcoded. When the platform is deployed in a region where only certain currencies are expected, the validation set is narrowed through configuration rather than code changes.

↑ Back to top · Next: Business Rules Validation →

---

📋 Business Rules Validation

Business rules validation checks semantic constraints — conditions that cannot be expressed as type or format checks:

DEFAULT_BUSINESS_RULES = [
    {
        "name": "positive_amount",
        "description": "Transaction amounts must be positive",
        "check": lambda tx: tx.amount > 0,
        "blocking": True,
        "severity": "error",
    },
    {
        "name": "amount_threshold",
        "description": "Transactions above the high-value threshold require additional context",
        "check": lambda tx: tx.amount <= HIGH_VALUE_THRESHOLD or tx.memo is not None,
        "blocking": False,
        "severity": "warning",
    },
    {
        "name": "future_date",
        "description": "Transaction dates must not be in the future",
        "check": lambda tx: tx.transaction_date <= date.today(),
        "blocking": True,
        "severity": "error",
    },
    {
        "name": "account_format",
        "description": "Account identifiers must match the configured format",
        "check": lambda tx: (
            tx.to_account is None or
            bool(re.match(ACCOUNT_FORMAT_PATTERN, tx.to_account))
        ),
        "blocking": False,
        "severity": "warning",
    },
]

The business rules are expressed as lambdas in the default configuration, but the BusinessRulesValidator also accepts rules defined in external configuration files. This enables compliance teams to add and modify validation rules without deploying code changes — a significant operational advantage for rapidly changing regulatory requirements.

A concrete example: when a new regulatory threshold was introduced requiring enhanced due diligence on cross-border transactions above a specific amount, the rule was added to the configuration file and took effect on the next pipeline run without a deployment.

↑ Back to top · Next: Statistical Validation →

---

📊 Statistical Validation

Statistical validation detects anomalies in batch-level aggregates:

class StatisticalValidator:
    def __init__(self, lookback_days: int = 30):
        self.lookback_days = lookback_days
    def validate_batch(
        self,
        transactions: list[Transaction],
        batch_id: str | None = None,
    ) -> ValidationResult:
        issues = []
        baseline = self._load_baseline(self.lookback_days)
        if baseline is None:
            return ValidationResult(issues=[])  # No baseline yet; first runs always pass
        batch_stats = self._compute_stats(transactions)
        # Check mean amount drift
        if baseline.amount_mean > 0:
            drift_ratio = abs(batch_stats.amount_mean - baseline.amount_mean) / baseline.amount_mean
            if drift_ratio > AMOUNT_DRIFT_THRESHOLD:
                issues.append(ValidationIssue(
                    record_index=None,
                    field="amount",
                    value=batch_stats.amount_mean,
                    rule="amount_mean_drift",
                    message=(
                        f"Batch mean amount {batch_stats.amount_mean:.2f} deviates "
                        f"{drift_ratio:.1%} from {self.lookback_days}-day baseline "
                        f"{baseline.amount_mean:.2f}"
                    ),
                    blocking=False,
                ))
        # Check batch size anomaly
        if baseline.batch_size_mean > 0:
            size_ratio = len(transactions) / baseline.batch_size_mean
            if size_ratio < BATCH_SIZE_MIN_RATIO or size_ratio > BATCH_SIZE_MAX_RATIO:
                issues.append(ValidationIssue(
                    record_index=None,
                    field="batch_size",
                    value=len(transactions),
                    rule="batch_size_anomaly",
                    message=(
                        f"Batch size {len(transactions)} is {size_ratio:.1f}x "
                        f"the {self.lookback_days}-day baseline mean of "
                        f"{baseline.batch_size_mean:.0f}"
                    ),
                    blocking=False,
                ))
        return ValidationResult(issues=issues)

Statistical issues are non-blocking by design. An anomalous batch is still ingested — the data is real — but the anomaly is flagged for investigation. The alert callback notifies the data team, and the validation result is recorded in pipeline_context for downstream lineage metadata.

The baseline computation uses a rolling window stored in the platform's metadata database. Each successfully processed batch contributes to the baseline for future batches, so validation accuracy improves over time as the system accumulates more history.

↑ Back to top · Next: Prometheus Metrics for Validation →

---

📈 Prometheus Metrics for Validation

Every validation failure is counted and labelled by rule type, enabling trend analysis in Grafana:

validation_issues_total = Counter(
    "pipeline_validation_issues_total",
    "Total validation issues by rule type and severity",
    ["rule", "severity", "blocking"],
)

validation_batch_duration_seconds = Histogram(
    "pipeline_validation_batch_duration_seconds",
    "Time to validate a batch",
    buckets=[0.1, 0.5, 1.0, 5.0, 10.0],
)

def _record_metrics(self, issues: list[ValidationIssue]) -> None:
    for issue in issues:
        validation_issues_total.labels(
            rule=issue.rule,
            severity=issue.severity,
            blocking=str(issue.blocking),
        ).inc()

The label cardinality is intentionally low: rule (bounded by the number of configured rules), severity (error or warning), blocking (True or False). This keeps metric storage manageable while providing the granularity needed to detect when a specific rule is firing at an elevated rate.

A Grafana alert on rate(pipeline_validation_issues_total{rule="valid_currency"}[5m]) > 0 immediately notifies the team when currency validation failures appear — which typically indicates a data source change rather than a one-off data quality issue.

↑ Back to top · Next: OpenLineage: The Provenance Standard →

---

🗂️ OpenLineage: The Provenance Standard

OpenLineage is an open standard for data lineage metadata. It defines a common vocabulary for describing what happened to data: which jobs ran, what datasets they consumed, what datasets they produced, and what facets (extended metadata) describe those events.

The core model has three concepts:

• 📦 Job: A repeatable process. In the platform, a PipelineRunner execution is a Job. Each job has a name and a namespace.
• ▶️ Run: A specific execution of a Job. A run has a runId (UUID), timestamps, and a state (START, COMPLETE, FAIL, ABORT).
• 🗃️ Dataset: A named collection of data. In the platform, the staged_data table, a MinIO object, and a Kafka topic are all Datasets. Each dataset has a name, a namespace, and optional facets.

Facets are typed extensions to the core model. SchemaDatasetFacet describes the dataset's schema. DataQualityMetricsInputDatasetFacet carries validation metrics. ColumnLineageDatasetFacet maps specific input columns to output columns. OwnershipDatasetFacet records who owns the data — required for GDPR data subject requests.

↑ Back to top · Next: The LineageRegistryStage →

---

🔗 The LineageRegistryStage

The LineageRegistryStage emits OpenLineage events to the OpenMetadata server at the end of each pipeline run:

class LineageRegistryStage(PipelineStage):
    def __init__(self, config: dict | None = None):
        self.config = config or {}
        self.openlineage_url = self.config.get(
            "openlineage_url",
            get_openlineage_url(),
        )
        self.namespace = self.config.get("namespace", "agentic-pfm")

    def name(self) -> str:
        return "LineageRegistryStage"

    def execute(
        self,
        data: tuple[list[Transaction], list[str]],
        pipeline_context: dict,
    ) -> tuple[list[Transaction], list[str]]:
        transactions, memos = data
        batch_id = pipeline_context.get("batch_id", str(uuid.uuid4()))
        run_id = pipeline_context.get("run_id", str(uuid.uuid4()))

        event = self._build_lineage_event(
            transactions=transactions,
            pipeline_context=pipeline_context,
            batch_id=batch_id,
            run_id=run_id,
        )

        try:
            self._emit_event(event)
            pipeline_context["lineage_registered"] = True
        except Exception as error:
            logger.warning("Lineage registration failed (non-blocking): %s", error)
            pipeline_context["lineage_registered"] = False

        return transactions, memos

    def _build_lineage_event(
        self,
        transactions: list[Transaction],
        pipeline_context: dict,
        batch_id: str,
        run_id: str,
    ) -> dict:
        return {
            "eventType": "COMPLETE",
            "eventTime": datetime.now(UTC).isoformat(),
            "run": {
                "runId": run_id,
                "facets": {
                    "processing": {
                        "_producer": "agentic-pfm-pipeline",
                        "_schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json",
                        "batch_id": batch_id,
                        "pipeline_version": pipeline_context.get("pipeline_version", "unknown"),
                        "record_count": len(transactions),
                    }
                },
            },
            "job": {
                "namespace": self.namespace,
                "name": "transaction_pipeline",
                "facets": {
                    "ownership": {
                        "_producer": "agentic-pfm-pipeline",
                        "_schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json",
                        "owners": [
                            {"name": pipeline_context.get("owner", "data-platform-team"), "type": "team"}
                        ],
                    }
                },
            },
            "inputs": [
                {
                    "namespace": self.namespace,
                    "name": "synthetic_transaction_generator",
                    "facets": {
                        "dataQualityMetrics": {
                            "_producer": "agentic-pfm-pipeline",
                            "_schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json",
                            "rowCount": len(transactions),
                            "validCount": len(transactions) - pipeline_context.get("validation_result", {}).get("blocking_issues", 0),
                            "invalidCount": pipeline_context.get("validation_result", {}).get("blocking_issues", 0),
                        }
                    },
                }
            ],
            "outputs": [
                {
                    "namespace": self.namespace,
                    "name": "staged_data",
                    "facets": {
                        "schema": {
                            "_producer": "agentic-pfm-pipeline",
                            "_schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json",
                            "fields": self._get_schema_facet(),
                        },
                        "ownership": {
                            "_producer": "agentic-pfm-pipeline",
                            "_schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json",
                            "owners": [
                                {"name": "data-platform-team", "type": "team"}
                            ],
                        },
                        "columnLineage": {
                            "_producer": "agentic-pfm-pipeline",
                            "_schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json",
                            "fields": {
                                "scrubbed_memo": {
                                    "inputFields": [
                                        {
                                            "namespace": self.namespace,
                                            "dataset": "synthetic_transaction_generator",
                                            "field": "memo",
                                        }
                                    ],
                                    "transformationDescription": "PII scrubbing via Presidio NER — named entities replaced with type labels",
                                    "transformationType": "IDENTITY",
                                }
                            },
                        },
                    },
                }
            ],
        }

The columnLineage facet records that scrubbed_memo was derived from memo through PII scrubbing. This is the data the GDPR right-to-explanation audit uses: when a data subject asks why their name is not in the analytics table, the lineage shows exactly which stage removed it, at which point in the pipeline.

↑ Back to top · Next: FAIR Data Principles in Practice →

---

🌐 FAIR Data Principles in Practice

The FAIR Data Principles (Findable, Accessible, Interoperable, Reusable) were originally formulated for scientific data, but they provide a useful framework for any data platform that needs to serve multiple consumers and persist data over years.

• 🔎 Findable: Every dataset in the platform is registered in OpenMetadata with a stable identifier, human-readable name, and searchable description. The OpenMetadata discovery UI lets data consumers find staged_data, see its schema, ownership, and quality metrics, and trace its lineage — without reading the pipeline code. The OwnershipDatasetFacet in the lineage event records the team responsible for the dataset. This is the pointer that GDPR data subject access requests use to identify who to contact.
• 🔓 Accessible: OpenMetadata provides a REST API for programmatic dataset access alongside the UI. All datasets follow a consistent access control model (the PostgreSQL role-based grants from the security article). A data consumer can discover a dataset in OpenMetadata and know immediately what credentials are needed to access it.
• 🔄 Interoperable: The platform uses standard vocabularies: ISO 4217 for currency codes, ISO 8601 for dates, OpenLineage for lineage events. Downstream consumers that understand these standards can consume the data without platform-specific adapters. The SchemaDatasetFacet uses the OpenLineage schema vocabulary, not a custom format — meaning any OpenLineage-compatible lineage visualiser (Marquez, OpenMetadata, Atlan) can render the platform's lineage without integration work.
• ♻️ Reusable: The staged_data table is designed for reuse, not just for the pipeline that produced it. The audit_metadata JSONB column carries the context needed to reuse data correctly: which batch produced this record, what governance decisions were made, what pipeline version ran, whether PII was present.

↑ Back to top · Next: GDPR Right to Explanation Through Lineage →

---

⚖️ GDPR Right to Explanation Through Lineage

GDPR Article 22 provides data subjects with the right to meaningful information about automated decision-making. For the financial platform, this means: if a transaction was flagged by the compliance audit agent, the data subject can request an explanation of why.

The lineage chain enables this explanation. Starting from a transaction in staged_data:

1. The audit_metadata.batch_id identifies the pipeline run that ingested it.
2. The OpenLineage event for that run identifies the transaction_pipeline job and the input source.
3. The columnLineage facet shows that scrubbed_memo was derived from memo through PII scrubbing.
4. The OPA decision log records the policy evaluation for the batch.
5. The compliance_decision in the governance stage output records the compliance agent's evaluation.

Each step in the chain is a structured artifact. The lineage query that reconstructs this chain:

def get_transaction_explanation(transaction_id: str) -> dict:
    """Retrieve the provenance chain for a transaction for GDPR Article 22."""
    with SessionLocal() as session:
        tx = session.query(StagedData).filter_by(id=transaction_id).first()
        if tx is None:
            return {"error": "Transaction not found"}

        batch_id = tx.audit_metadata.get("batch_id")
        governance_decision = tx.audit_metadata.get("governance_decision")
        pipeline_version = tx.audit_metadata.get("pipeline_version")

    lineage_event = fetch_lineage_event(batch_id)
    opa_decision = fetch_opa_decision(batch_id)

    return {
        "transaction_id": transaction_id,
        "processed_in_batch": batch_id,
        "pipeline_version": pipeline_version,
        "governance_decision": governance_decision,
        "opa_policy_evaluation": {
            "decision": opa_decision.get("result", {}).get("allow"),
            "reasons": opa_decision.get("result", {}).get("reasons", []),
        },
        "data_transformations": lineage_event.get("outputs", [{}])[0]
            .get("facets", {})
            .get("columnLineage", {})
            .get("fields", {}),
    }

This function returns everything needed to answer a data subject's explanation request: which pipeline processed their data, what version, what governance decision was reached, which OPA policies were evaluated, and what transformations were applied. The response is machine-readable for the data protection officer's case management system and human-readable for the data subject.

↑ Back to top · Next: OpenMetadata Registration (Non-Blocking) →

---

🗄️ OpenMetadata Registration (Non-Blocking)

The platform registers datasets in OpenMetadata as part of the governance stage. Registration is non-blocking: if OpenMetadata is unavailable, the pipeline continues. The data is still ingested; it just lacks the catalog entry until the next successful run.

class OpenMetadataRegistrationStage(PipelineStage):
    def __init__(self, config: dict | None = None):
        self.config = config or {}
        self.client = OpenMetadataClient(
            host=self.config.get("openmetadata_host", get_openmetadata_url()),
        )

    def name(self) -> str:
        return "OpenMetadataRegistrationStage"

    def execute(
        self,
        data: tuple[list[Transaction], list[str]],
        pipeline_context: dict,
    ) -> tuple[list[Transaction], list[str]]:
        try:
            self.client.upsert_table(
                database="agentic_pfm",
                schema="public",
                table="staged_data",
                columns=get_staged_data_columns(),
                owner=pipeline_context.get("owner", "data-platform-team"),
                description=(
                    "Ingested and governed financial transactions. "
                    "Contains scrubbed memos (original_memo retained under SOX obligation). "
                    "Lineage tracked via OpenLineage."
                ),
                tags=["financial", "pii-present", "sox-retained", "gdpr-controlled"],
            )
            pipeline_context["openmetadata_registered"] = True
        except Exception as error:
            logger.warning("OpenMetadata registration skipped: %s", error)
            pipeline_context["openmetadata_registered"] = False

        return data

The upsert_table call makes re-registration safe: it updates the existing catalog entry rather than creating a duplicate. This makes registration idempotent — the pipeline can call it on every run without accumulating stale entries.

↑ Back to top · Next: Lineage Visualisation in OpenMetadata →

---

🖥️ Lineage Visualisation in OpenMetadata

Once lineage events are flowing to OpenMetadata, the lineage graph becomes navigable in the UI:

• From staged_data, clicking upstream shows the transaction_pipeline job and the source generator.
• Clicking the pipeline job shows all runs, with timestamps, record counts, and validation metrics from the data quality facets.
• The column lineage view shows scrubbed_memo ← [PII scrubbing] ← memo, making the transformation history visible without reading the pipeline code.

For a compliance auditor, the lineage graph is the primary artefact for demonstrating that data handling controls work as documented. They can navigate from any row in staged_data back through the pipeline to the ingestion event, verifying at each step that the appropriate controls ran.

↑ Back to top · Next: Key Takeaways →

---

---

❓ Frequently Asked Questions

Common questions about data validation, OpenLineage integration, and GDPR data provenance answered from real-world implementation experience.

What is the difference between schema validation, business rules validation, and statistical validation?

Schema validation checks structural correctness — is currency a string matching ISO 4217? Is amount a non-negative number? Business rules validation checks semantic correctness — a valid number like -99999 is still an invalid transaction amount; a valid three-character string like "ZZZ" is still an unrecognised currency. Statistical validation checks batch-level anomalies against a rolling baseline — a 400% jump in mean transaction amount may indicate a data source problem even if every individual record passes the other two layers.

How does OpenLineage work and what data does it capture?

OpenLineage is an open standard for lineage metadata built around three concepts: a Job (the pipeline), a Run (a specific execution with a UUID), and Datasets (input and output data stores). Each run emits a COMPLETE event that carries facets: SchemaDatasetFacet (column names and types), DataQualityMetricsInputDatasetFacet (row counts, valid/invalid split), and ColumnLineageDatasetFacet (which input column produced which output column, and via what transformation). Any OpenLineage-compatible tool — Marquez, OpenMetadata, Atlan — can consume these events without custom integration.

How does data lineage satisfy GDPR right-to-explanation requirements?

The GDPR Article 22 right to meaningful information about automated decisions is satisfied through a provenance chain: the audit_metadata.batch_id identifies the pipeline run; the OpenLineage event for that run shows the input source and job; the columnLineage facet shows that scrubbed_memo was derived from memo through PII scrubbing; the OPA decision log records the governance policy evaluation. The get_transaction_explanation() function assembles this chain into a structured response for data protection officers and data subjects.

Why should OpenMetadata registration be non-blocking in a data pipeline?

OpenMetadata registration is a metadata cataloging operation — it records that a dataset exists and updates its schema, owner, and tags. If the catalog is temporarily unavailable and registration blocks the pipeline, data ingestion stops even though the data itself is valid and the database write would succeed. The pipeline uses upsert_table (idempotent) rather than create, so the next successful run will update the catalog entry. A metadata failure should never prevent data from reaching its consumers — only a data quality failure should.

↑ Back to top · Next: Key Takeaways →

---

---