CI/CD & Production Workflows¶

Hardcoding comparison rules within application logic creates maintenance bottlenecks in production environments. This notebook demonstrates how to decouple Policy (validation rules) from Execution (pipeline code) using Veridelta's YAML configuration.

1. The GitOps Validation Pattern¶

Veridelta enables a declarative, GitOps-driven workflow for data pipeline validation:

1. Store: Maintain veridelta.yaml in version control alongside data models.
2. Execute: Trigger a generic validation script during the CI/CD pipeline.
3. Update: Modify comparison tolerances via Pull Requests without altering core Python application code.

In [ ]:

import pathlib

import polars as pl

from veridelta import DataIngestor, DiffEngine, load_config

# 1. Mock production data (e.g., Nightly User LTV Pipeline)
src_df = pl.DataFrame(
    {
        "user_id": ["U-100", "U-101"],
        "ltv_score": [850.50, 420.00],
        "cohort": ["Q1_Active", "Q2_Churned"],
    }
)

# Simulate downstream drift ($0.05 drift; lowercase strings)
tgt_df = pl.DataFrame(
    {
        "user_id": ["U-100", "U-101"],
        "ltv_score": [850.55, 420.00],
        "cohort": ["q1_active", "q2_churned"],
    }
)

# Simulate remote storage artifacts (S3/GCS) via local disk
src_df.write_parquet("prod_source.parquet")
tgt_df.write_parquet("prod_target.parquet")

# 2. Define the declarative policy (veridelta.yaml)
yaml_content = """
primary_keys: ["user_id"]
default_absolute_tolerance: 0.10
rules:
  - column_names: ["cohort"]
    case_insensitive: true
source:
  path: "prod_source.parquet"
  format: "parquet"
target:
  path: "prod_target.parquet"
  format: "parquet"
"""
pathlib.Path("veridelta.yaml").write_text(yaml_content)
print("Production config and mock artifacts generated.")

# Output:
# Production config and mock artifacts generated.

import pathlib import polars as pl from veridelta import DataIngestor, DiffEngine, load_config # 1. Mock production data (e.g., Nightly User LTV Pipeline) src_df = pl.DataFrame( { "user_id": ["U-100", "U-101"], "ltv_score": [850.50, 420.00], "cohort": ["Q1_Active", "Q2_Churned"], } ) # Simulate downstream drift ($0.05 drift; lowercase strings) tgt_df = pl.DataFrame( { "user_id": ["U-100", "U-101"], "ltv_score": [850.55, 420.00], "cohort": ["q1_active", "q2_churned"], } ) # Simulate remote storage artifacts (S3/GCS) via local disk src_df.write_parquet("prod_source.parquet") tgt_df.write_parquet("prod_target.parquet") # 2. Define the declarative policy (veridelta.yaml) yaml_content = """ primary_keys: ["user_id"] default_absolute_tolerance: 0.10 rules: - column_names: ["cohort"] case_insensitive: true source: path: "prod_source.parquet" format: "parquet" target: path: "prod_target.parquet" format: "parquet" """ pathlib.Path("veridelta.yaml").write_text(yaml_content) print("Production config and mock artifacts generated.") # Output: # Production config and mock artifacts generated.

2. Generic Execution Script¶

Pipeline orchestrators (e.g., GitHub Actions, Airflow, dbt) utilize a generic execution script. This script remains fully agnostic to schemas, tolerances, and formats, delegating all validation logic to the YAML configuration.

In [ ]:

def run_pipeline_validation(config_path: str) -> None:
    """Load config and execute the semantic diff pipeline.

    Args:
        config_path: Path to the veridelta.yaml file.
    """
    diff_cfg, src_cfg, tgt_cfg = load_config(config_path)

    ingestor = DataIngestor(diff_cfg, src_cfg, tgt_cfg)
    source_df, target_df = ingestor.get_dataframes()

    engine = DiffEngine(diff_cfg, source_df, target_df)
    summary = engine.run()

    print(f"{' CI/CD PIPELINE VALIDATION ':=^50}")
    if summary.is_match:
        print("STATUS: PASSED")
    else:
        print(f"STATUS: FAILED ({summary.changed_count} regressions found)")
        # In production CI/CD, raise an exception to block the deployment:
        # raise AssertionError("Data regression detected!")
    print("=" * 50)


run_pipeline_validation("veridelta.yaml")

# Output:
# =========== CI/CD PIPELINE VALIDATION ===========
# STATUS: PASSED
# =================================================

def run_pipeline_validation(config_path: str) -> None: """Load config and execute the semantic diff pipeline. Args: config_path: Path to the veridelta.yaml file. """ diff_cfg, src_cfg, tgt_cfg = load_config(config_path) ingestor = DataIngestor(diff_cfg, src_cfg, tgt_cfg) source_df, target_df = ingestor.get_dataframes() engine = DiffEngine(diff_cfg, source_df, target_df) summary = engine.run() print(f"{' CI/CD PIPELINE VALIDATION ':=^50}") if summary.is_match: print("STATUS: PASSED") else: print(f"STATUS: FAILED ({summary.changed_count} regressions found)") # In production CI/CD, raise an exception to block the deployment: # raise AssertionError("Data regression detected!") print("=" * 50) run_pipeline_validation("veridelta.yaml") # Output: # =========== CI/CD PIPELINE VALIDATION =========== # STATUS: PASSED # =================================================

Advanced: Environment Overrides¶

During environment promotion (e.g., Staging to Production), the SourceConfig.options block accepts dynamic cloud credentials or environment-specific path overrides. This ensures semantic validation rules remain strictly identical across all deployment tiers.

---

(Run the cell below to clean up the temporary files generated by this notebook)

In [ ]:

# Remove local artifacts (housekeeping)
for f in ["prod_source.parquet", "prod_target.parquet", "veridelta.yaml"]:
    pathlib.Path(f).unlink(missing_ok=True)

# Remove local artifacts (housekeeping) for f in ["prod_source.parquet", "prod_target.parquet", "veridelta.yaml"]: pathlib.Path(f).unlink(missing_ok=True)