Metadata Standards, Schema Governance & Data Provenance Contracts: Best Practices for Distributed Analytics Systems¶

# validation/great_expectations/roads_v2_suite.py
import great_expectations as ge
from great_expectations.dataset import PandasDataset

def create_roads_v2_expectation_suite():
    """Create expectation suite for roads-v2 dataset"""
    suite = ge.ExpectationSuite("roads_v2")

    # Schema expectations
    suite.expect_column_to_exist("road_id")
    suite.expect_column_to_exist("road_name")
    suite.expect_column_to_exist("risk_level")
    suite.expect_column_to_exist("geometry")

    # Type expectations
    suite.expect_column_values_to_be_of_type("road_id", "int64")
    suite.expect_column_values_to_be_of_type("road_name", "object")
    suite.expect_column_values_to_be_of_type("risk_level", "object")

    # Constraint expectations
    suite.expect_column_values_to_be_unique("road_id")
    suite.expect_column_values_to_not_be_null("road_id")
    suite.expect_column_values_to_not_be_null("risk_level")
    suite.expect_column_values_to_be_in_set(
        "risk_level",
        ["low", "medium", "high"]
    )

    # Quality expectations
    suite.expect_column_values_to_not_be_null("road_name", mostly=0.98)
    suite.expect_column_value_lengths_to_be_between("road_name", 1, 255)

    return suite

# validation/polars/validate_roads_v2.py
import polars as pl
from typing import Dict, Any

def validate_roads_v2_schema(df: pl.DataFrame, contract: Dict[str, Any]) -> bool:
    """Validate DataFrame against contract schema"""
    schema = contract["schema"]

    # Check required fields
    required_fields = schema["requiredFields"]
    missing_fields = set(required_fields) - set(df.columns)
    if missing_fields:
        raise ValueError(f"Missing required fields: {missing_fields}")

    # Check field types
    for field in schema["fields"]:
        field_name = field["name"]
        if field_name not in df.columns:
            continue

        expected_type = field["type"]["parquet"]
        actual_type = str(df[field_name].dtype)

        # Type mapping
        type_map = {
            "int64": "Int64",
            "string": "Utf8",
            "float64": "Float64",
            "bool": "Boolean",
            "timestamp[us, tz=UTC]": "Datetime"
        }

        if type_map.get(expected_type) != actual_type:
            raise ValueError(
                f"Field {field_name}: expected {expected_type}, got {actual_type}"
            )

        # Check constraints
        if "constraints" in field:
            constraints = field["constraints"]

            # Check allowed values
            if "allowedValues" in constraints:
                allowed = set(constraints["allowedValues"])
                actual = set(df[field_name].unique().to_list())
                invalid = actual - allowed
                if invalid:
                    raise ValueError(
                        f"Field {field_name}: invalid values {invalid}"
                    )

            # Check nullability
            if not field.get("nullable", False):
                null_count = df[field_name].null_count()
                if null_count > 0:
                    raise ValueError(
                        f"Field {field_name}: {null_count} null values found"
                    )

    return True

# prefect/tasks/validate_contract.py
from prefect import task
import polars as pl
from validation.polars.validate_roads_v2 import validate_roads_v2_schema
import yaml

@task
def validate_data_contract(
    data_path: str,
    contract_path: str
) -> bool:
    """Validate data against contract"""
    # Load contract
    with open(contract_path, 'r') as f:
        contract = yaml.safe_load(f)

    # Load data
    df = pl.read_parquet(data_path)

    # Validate
    try:
        validate_roads_v2_schema(df, contract)
        return True
    except ValueError as e:
        raise ValueError(f"Contract validation failed: {e}")

# mlflow/model_metadata.py
import mlflow
import mlflow.sklearn
from datetime import datetime

def log_model_with_schema(
    model,
    training_data_schema_version: str,
    contract_version: str
):
    """Log MLflow model with schema version metadata"""
    with mlflow.start_run():
        # Log model
        mlflow.sklearn.log_model(model, "model")

        # Log schema version metadata
        mlflow.log_param("training_data_schema_version", training_data_schema_version)
        mlflow.log_param("contract_version", contract_version)
        mlflow.log_param("schema_version", "2.0.0")

        # Log contract reference
        mlflow.log_artifact("contracts/ml-risk-model-input-v1.yaml")
        mlflow.log_artifact("contracts/ml-risk-model-output-v1.yaml")

        # Log provenance
        mlflow.log_param("training_data", "roads-v2")
        mlflow.log_param("training_data_version", "2.1.0")
        mlflow.log_param("code_version", "git://github.com/org/ml-models@abc123")

# onnx/schema_verification.py
import onnxruntime as ort
import numpy as np
from validation.polars.validate_roads_v2 import validate_roads_v2_schema

def verify_input_schema(
    input_data: np.ndarray,
    model_path: str,
    contract_path: str
) -> bool:
    """Verify input data matches model's expected schema"""
    # Load model
    session = ort.InferenceSession(model_path)

    # Get expected input schema from model metadata
    input_meta = session.get_inputs()[0]
    expected_shape = input_meta.shape
    expected_type = input_meta.type

    # Verify input matches
    if input_data.shape != tuple(expected_shape):
        raise ValueError(
            f"Input shape mismatch: expected {expected_shape}, got {input_data.shape}"
        )

    # Load and validate against contract
    with open(contract_path, 'r') as f:
        contract = yaml.safe_load(f)

    # Validate data against contract
    validate_roads_v2_schema(input_data, contract)

    return True

# provenance/postgres_lineage.py
import psycopg2
from datetime import datetime
import uuid

def record_lineage_event(
    inputs: list,
    outputs: list,
    transformation: dict,
    user: str,
    cluster: str,
    node: str
):
    """Record lineage event in Postgres"""
    conn = psycopg2.connect("postgresql://user:pass@localhost/metadata")
    cur = conn.cursor()

    event_id = uuid.uuid4()

    # Insert event
    cur.execute("""
        INSERT INTO provenance_events (
            event_id, timestamp, event_type,
            transformation_code, transformation_code_hash,
            environment_hash, executed_by, execution_id,
            user_email, cluster_name, node_name
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """, (
        event_id,
        datetime.utcnow(),
        "transformation",
        transformation["code"],
        transformation["codeHash"],
        transformation["environment"],
        transformation["executedBy"],
        transformation["executionId"],
        user,
        cluster,
        node
    ))

    # Insert input relationships
    for input_dataset in inputs:
        cur.execute("""
            INSERT INTO provenance_relationships (
                event_id, relationship_type,
                dataset_id, dataset_version,
                location, checksum
            ) VALUES (%s, %s, %s, %s, %s, %s)
        """, (
            event_id,
            "input",
            input_dataset["datasetId"],
            input_dataset["version"],
            input_dataset["location"],
            input_dataset["checksum"]
        ))

    # Insert output relationships
    for output_dataset in outputs:
        cur.execute("""
            INSERT INTO provenance_relationships (
                event_id, relationship_type,
                dataset_id, dataset_version,
                location, checksum
            ) VALUES (%s, %s, %s, %s, %s, %s)
        """, (
            event_id,
            "output",
            output_dataset["datasetId"],
            output_dataset["version"],
            output_dataset["location"],
            output_dataset["checksum"]
        ))

    conn.commit()
    cur.close()
    conn.close()

# provenance/parquet_lineage.py
import pyarrow.parquet as pq
import pyarrow as pa
import json

def embed_lineage_in_parquet(
    parquet_path: str,
    lineage_event: dict
):
    """Embed lineage metadata in Parquet file"""
    # Read existing Parquet file
    table = pq.read_table(parquet_path)

    # Add lineage metadata to custom metadata
    custom_metadata = {
        "lineage": json.dumps(lineage_event),
        "provenance": json.dumps({
            "inputs": lineage_event["inputs"],
            "transformation": lineage_event["transformation"]
        })
    }

    # Write with custom metadata
    pq.write_table(
        table,
        parquet_path,
        metadata_collector=lambda metadata: metadata.update(custom_metadata)
    )

# mlflow/provenance.py
import mlflow

def log_provenance(
    training_data: str,
    training_data_version: str,
    code_version: str,
    environment_hash: str
):
    """Log provenance to MLflow"""
    with mlflow.start_run():
        # Log data provenance
        mlflow.log_param("training_data", training_data)
        mlflow.log_param("training_data_version", training_data_version)
        mlflow.log_param("code_version", code_version)
        mlflow.log_param("environment_hash", environment_hash)

        # Log parent runs (if applicable)
        # mlflow.set_tag("mlflow.parentRunId", parent_run_id)

# prefect/tasks/etl_with_lineage.py
from prefect import task, flow
from provenance.postgres_lineage import record_lineage_event

@task
def transform_roads(input_path: str, output_path: str):
    """Transform roads data with lineage tracking"""
    # Load input
    df = pl.read_parquet(input_path)

    # Transform
    df_transformed = df.with_columns([
        pl.col("risk_level").cast(pl.Utf8).alias("risk_level")
    ])

    # Write output
    df_transformed.write_parquet(output_path)

    # Record lineage
    record_lineage_event(
        inputs=[{
            "datasetId": "osm-roads-v1",
            "version": "1.5.0",
            "location": input_path,
            "checksum": compute_checksum(input_path)
        }],
        outputs=[{
            "datasetId": "roads-v2",
            "version": "2.1.0",
            "location": output_path,
            "checksum": compute_checksum(output_path)
        }],
        transformation={
            "code": "git://github.com/org/etl-pipelines@abc123",
            "codeHash": compute_code_hash(),
            "environment": compute_environment_hash(),
            "executedBy": "prefect-job-789",
            "executionId": prefect.context.get("run_id")
        },
        user="data-eng@example.com",
        cluster="rke2-prod",
        node="worker-01"
    )

    return output_path

@flow
def etl_roads_flow():
    """ETL flow with lineage tracking"""
    input_path = "s3://data-lake/datasets/osm-roads/v1/"
    output_path = "s3://data-lake/datasets/roads/v2/"

    transform_roads(input_path, output_path)