🚀 CI/CD for Analytics Pipelines Lab¶

# Configure Synapse workspace Git integration via Azure Portal or ARM template
{
  "type": "Microsoft.Synapse/workspaces",
  "properties": {
    "workspaceRepositoryConfiguration": {
      "type": "GitHub",
      "accountName": "your-github-org",
      "repositoryName": "synapse-analytics",
      "collaborationBranch": "main",
      "rootFolder": "/synapse",
      "tenantId": "YOUR_TENANT_ID"
    }
  }
}

# tests/unit/test_notebook_functions.py
"""Unit tests for Synapse notebook functions"""

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

@pytest.fixture(scope="session")
def spark():
    """Create Spark session for testing"""
    return SparkSession.builder \
        .appName("NotebookUnitTests") \
        .master("local[2]") \
        .getOrCreate()

@pytest.fixture
def sample_data(spark):
    """Create sample test data"""
    schema = StructType([
        StructField("customer_id", StringType(), True),
        StructField("transaction_amount", DoubleType(), True),
        StructField("category", StringType(), True)
    ])

    data = [
        ("CUST001", 100.50, "Electronics"),
        ("CUST002", 50.25, "Clothing"),
        ("CUST003", 200.00, "Electronics"),
    ]

    return spark.createDataFrame(data, schema)

def test_data_transformation(spark, sample_data):
    """Test data transformation logic"""
    from synapse.notebook.transformation import apply_business_rules

    result = apply_business_rules(sample_data)

    assert result.count() == 3
    assert "discount_applied" in result.columns
    assert result.filter("discount_applied = true").count() > 0

def test_aggregation_logic(spark, sample_data):
    """Test aggregation calculations"""
    from synapse.notebook.analytics import calculate_customer_metrics

    metrics = calculate_customer_metrics(sample_data)

    assert metrics.count() == 3
    assert "total_spend" in metrics.columns

    electronics_spend = metrics.filter("category = 'Electronics'") \
        .agg({"total_spend": "sum"}).collect()[0][0]

    assert electronics_spend == 300.50

def test_data_quality_checks(spark, sample_data):
    """Test data quality validation"""
    from synapse.notebook.validation import validate_data_quality

    validation_result = validate_data_quality(sample_data)

    assert validation_result["is_valid"] == True
    assert validation_result["null_count"] == 0
    assert validation_result["duplicate_count"] == 0

# tests/data-quality/test_data_quality.py
"""Data quality tests using Great Expectations"""

import great_expectations as gx
from great_expectations.data_context import DataContext

def test_transaction_data_quality():
    """Validate transaction data quality"""

    # Initialize Great Expectations
    context = DataContext("./great_expectations")

    # Create expectation suite
    suite = context.create_expectation_suite(
        "transaction_quality_suite",
        overwrite_existing=True
    )

    # Add expectations
    batch = context.get_batch({
        "path": "abfss://data@storage.dfs.core.windows.net/transactions/",
        "reader_method": "read_delta"
    }, suite)

    # Expected column presence
    batch.expect_table_columns_to_match_ordered_list([
        "transaction_id", "customer_id", "product_id",
        "amount", "quantity", "transaction_date"
    ])

    # No null values in key columns
    batch.expect_column_values_to_not_be_null("transaction_id")
    batch.expect_column_values_to_not_be_null("customer_id")
    batch.expect_column_values_to_not_be_null("amount")

    # Value constraints
    batch.expect_column_values_to_be_between("amount", min_value=0, max_value=100000)
    batch.expect_column_values_to_be_between("quantity", min_value=1, max_value=1000)

    # Format validation
    batch.expect_column_values_to_match_regex(
        "transaction_id",
        regex="^TXN-[0-9]{6}$"
    )

    # Uniqueness check
    batch.expect_column_values_to_be_unique("transaction_id")

    # Save suite
    batch.save_expectation_suite()

    # Run validation
    results = context.run_checkpoint(
        checkpoint_name="data_quality_checkpoint",
        batch_request={
            "datasource_name": "synapse_datasource",
            "data_connector_name": "default_runtime_data_connector",
            "data_asset_name": "transactions"
        }
    )

    assert results["success"], "Data quality validation failed"

def test_customer_data_completeness():
    """Validate customer data completeness and accuracy"""

    context = DataContext("./great_expectations")

    batch = context.get_batch({
        "path": "abfss://data@storage.dfs.core.windows.net/customers/",
        "reader_method": "read_delta"
    })

    # Email format validation
    batch.expect_column_values_to_match_regex(
        "email",
        regex="^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"
    )

    # Phone number validation
    batch.expect_column_values_to_match_regex(
        "phone",
        regex="^\\+?[1-9]\\d{1,14}$"
    )

    # Age constraints
    batch.expect_column_values_to_be_between("age", min_value=18, max_value=120)

    results = batch.validate()
    assert results["success"], "Customer data validation failed"

# tests/integration/test_pipeline_integration.py
"""Integration tests for Synapse pipelines"""

import pytest
from azure.identity import DefaultAzureCredential
from azure.synapse.artifacts import ArtifactsClient
from azure.synapse.artifacts.models import PipelineRunResponse
import time

@pytest.fixture
def synapse_client():
    """Create Synapse artifacts client"""
    credential = DefaultAzureCredential()
    endpoint = f"https://{os.getenv('SYNAPSE_WORKSPACE')}.dev.azuresynapse.net"
    return ArtifactsClient(credential, endpoint)

def test_ingestion_pipeline_execution(synapse_client):
    """Test data ingestion pipeline end-to-end"""

    # Trigger pipeline
    pipeline_name = "DataIngestionPipeline"
    run_response = synapse_client.pipeline_run.run_pipeline(
        pipeline_name,
        parameters={
            "sourcePath": "abfss://source@storage.dfs.core.windows.net/test-data/",
            "targetPath": "abfss://target@storage.dfs.core.windows.net/ingested/"
        }
    )

    run_id = run_response.run_id
    print(f"Pipeline run started: {run_id}")

    # Wait for completion (with timeout)
    timeout = 600  # 10 minutes
    start_time = time.time()

    while True:
        status = synapse_client.pipeline_run.get_pipeline_run(run_id)

        if status.status in ['Succeeded', 'Failed', 'Cancelled']:
            break

        if time.time() - start_time > timeout:
            pytest.fail("Pipeline execution timeout")

        time.sleep(30)

    # Assert success
    assert status.status == 'Succeeded', f"Pipeline failed: {status.message}"

    # Validate output data exists
    from azure.storage.filedatalake import DataLakeServiceClient

    credential = DefaultAzureCredential()
    service_client = DataLakeServiceClient(
        account_url="https://storage.dfs.core.windows.net",
        credential=credential
    )

    file_system = service_client.get_file_system_client("target")
    paths = file_system.get_paths(path="ingested/")

    assert any(paths), "No output files found"

def test_transformation_pipeline_data_quality(synapse_client):
    """Test transformation pipeline produces quality data"""

    pipeline_name = "DataTransformationPipeline"
    run_response = synapse_client.pipeline_run.run_pipeline(pipeline_name)

    # Wait for completion
    run_id = run_response.run_id
    status = wait_for_pipeline_completion(synapse_client, run_id)

    assert status.status == 'Succeeded'

    # Validate transformed data quality
    from pyspark.sql import SparkSession

    spark = SparkSession.builder.getOrCreate()
    transformed_df = spark.read.format("delta").load(
        "abfss://target@storage.dfs.core.windows.net/transformed/"
    )

    # Check row count
    assert transformed_df.count() > 0, "No transformed data found"

    # Check for null values in key columns
    null_counts = transformed_df.select([
        sum(col(c).isNull().cast("int")).alias(c)
        for c in transformed_df.columns
    ]).collect()[0]

    assert all(count == 0 for count in null_counts), "Null values found in output"

def wait_for_pipeline_completion(client, run_id, timeout=600):
    """Helper function to wait for pipeline completion"""
    start_time = time.time()

    while True:
        status = client.pipeline_run.get_pipeline_run(run_id)

        if status.status in ['Succeeded', 'Failed', 'Cancelled']:
            return status

        if time.time() - start_time > timeout:
            raise TimeoutError("Pipeline execution timeout")

        time.sleep(30)

# monitoring/deployment-monitor.py
"""Monitor deployment health and trigger rollback if needed"""

from azure.monitor.query import LogsQueryClient, MetricsQueryClient
from azure.identity import DefaultAzureCredential
from datetime import datetime, timedelta

def monitor_deployment_health(workspace_name, resource_group):
    """Monitor Synapse workspace health after deployment"""

    credential = DefaultAzureCredential()
    logs_client = LogsQueryClient(credential)

    # Query for errors in last 15 minutes
    query = f"""
    SynapseIntegrationPipelineRuns
    | where TimeGenerated > ago(15m)
    | where Status == "Failed"
    | summarize FailureCount = count() by PipelineName
    | where FailureCount > 3
    """

    end_time = datetime.now()
    start_time = end_time - timedelta(minutes=15)

    response = logs_client.query_workspace(
        workspace_id=os.getenv("LOG_ANALYTICS_WORKSPACE_ID"),
        query=query,
        timespan=(start_time, end_time)
    )

    failures = []
    for table in response.tables:
        for row in table.rows:
            failures.append({
                "pipeline": row[0],
                "failure_count": row[1]
            })

    if failures:
        print("⚠️ Deployment health check FAILED")
        print(f"Failed pipelines: {failures}")
        return False
    else:
        print("✅ Deployment health check PASSED")
        return True

def rollback_deployment(workspace_name, snapshot_name):
    """Rollback to previous deployment snapshot"""

    print(f"🔄 Rolling back to snapshot: {snapshot_name}")

    # Download snapshot from storage
    from azure.storage.blob import BlobServiceClient

    credential = DefaultAzureCredential()
    blob_service = BlobServiceClient(
        account_url="https://storage.blob.core.windows.net",
        credential=credential
    )

    container_client = blob_service.get_container_client("deployment-snapshots")
    blob_client = container_client.get_blob_client(f"{snapshot_name}.tar.gz")

    # Download and extract snapshot
    with open("rollback-snapshot.tar.gz", "wb") as f:
        f.write(blob_client.download_blob().readall())

    # Extract and deploy
    import tarfile
    with tarfile.open("rollback-snapshot.tar.gz", "r:gz") as tar:
        tar.extractall("./rollback")

    # Deploy snapshot artifacts
    import subprocess
    subprocess.run([
        "pwsh", "infrastructure/scripts/deploy-artifacts.ps1",
        "-WorkspaceName", workspace_name,
        "-ResourceGroup", os.getenv("RESOURCE_GROUP"),
        "-ArtifactsPath", "./rollback/synapse"
    ])

    print("✅ Rollback completed")

# Monitor deployment
if __name__ == "__main__":
    workspace = os.getenv("SYNAPSE_WORKSPACE")
    resource_group = os.getenv("RESOURCE_GROUP")

    is_healthy = monitor_deployment_health(workspace, resource_group)

    if not is_healthy:
        latest_snapshot = os.getenv("LAST_KNOWN_GOOD_SNAPSHOT")
        rollback_deployment(workspace, latest_snapshot)

def validate_cicd_implementation():
    """Validate CI/CD implementation"""

    checklist = {
        "Version Control": False,
        "CI Pipeline": False,
        "Automated Tests": False,
        "CD Pipeline": False,
        "Monitoring": False
    }

    # Check Git repository
    if os.path.exists(".git"):
        checklist["Version Control"] = True

    # Check CI/CD pipeline files
    if os.path.exists(".github/workflows/ci-pipeline.yml"):
        checklist["CI Pipeline"] = True

    if os.path.exists(".github/workflows/cd-pipeline.yml"):
        checklist["CD Pipeline"] = True

    # Check test directory
    if os.path.exists("tests/unit") and os.path.exists("tests/integration"):
        checklist["Automated Tests"] = True

    # Check monitoring setup
    if os.path.exists("monitoring/deployment-monitor.py"):
        checklist["Monitoring"] = True

    print("CI/CD Implementation Checklist:")
    for item, status in checklist.items():
        print(f"{'✅' if status else '❌'} {item}")

    return all(checklist.values())

# Run validation
validate_cicd_implementation()