Great Expectations Data Quality Validation using Azure Functions with Azure AI foundry- Part3.

8 min readSep 4, 2025

Introduction

This article presents a comprehensive enterprise-grade data quality validation solution that combines the power of Great Expectations for data validation, Azure Functions for serverless computing, and Azure AI Foundry for intelligent insights. The solution provides automated data quality assessment across multiple data sources including Databricks Unity Catalog, SQL Server, Web APIs, and web scraping.

Solution Overview

This comprehensive solution demonstrates how to build an enterprise-grade data quality validation platform that seamlessly integrates Great Expectations for robust data validation using Azure Functions and getting insights from Azure AI Foundry for intelligent insights as MCP (Model Context Protocol). The platform provides automated data quality assessment across multiple data sources while delivering AI-powered recommendations.

Integrates with Azure AI Foundry and Azure ML for intelligent anomaly detection, predictive analysis, and automated recommendations.

(One of the question folks may ask rather than using Azure AI foundry we could have used Databricks Genie as well, again my intention was to use Azure AI foundry because to check Response differences between Genie vs Azure AI foundry and then cost per se). and then when you use it with Azure Function you can pass different parameters and ask Questions on the fly as for end user I can avoid giving direct access to Databricks rather they can run FunctionApp API and get results on the fly.

Architecture & Components

Implementation Steps

Here are the steps to follow to run Data Quality using Great Expectations using Azure Function App code and then Interacting with Azure AI foundry as MCP to get suggestions.

Step 1: Setup & Prerequisites

# Install required dependencies
pip install azure-functions==1.15.0
pip install great-expectations==1.0.0
pip install databricks-sql-connector==3.1.2
pip install azure-identity==1.15.0
pip install azure-ai-projects==1.0.0b1
pip install pandas==2.1.4
pip install scikit-learn==1.3.2

Step 2: Setup new Azure Function

First step is to Create your Azure Function project with the following structure.

DatabricksConnector/
├── __init__.py         
├── function.json        
├── requirements.txt    
├── host.json           
├── local.settings.json 
└── great_expectations/  
    ├── great_expectations.yml
    ├── checkpoints/
    └── expectations/

Step 3: Core Function Implementation

Key Implementation: The main Azure Function handles HTTP requests, processes data from multiple sources, runs Great Expectations validations, and integrates with Azure AI Foundry.

import azure.functions as func
import pandas as pd
import great_expectations as gx
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
import os
import logging

def main(req: func.HttpRequest) -> func.HttpResponse:
    """Main Azure Function entry point"""
    try:
        # Parse request body
        req_body = req.get_json()
        user_prompt = req_body.get("prompt", "")        # Step 1: Load data from configured source
        if "[DATAFRAME_PLACEHOLDER]" in user_prompt:
            df, table_name = get_data_source()
            preview = df.head(10).to_string(index=False)
            user_prompt = user_prompt.replace("[DATAFRAME_PLACEHOLDER]", preview)        # Step 2: Run Great Expectations validation
        if "[GE_SUMMARY_PLACEHOLDER]" in user_prompt:
            ge_summary = run_great_expectations_validation()
            user_prompt = user_prompt.replace("[GE_SUMMARY_PLACEHOLDER]", ge_summary)        # Step 3: Perform predictive analysis
        if "[PREDICTIVE_ANALYSIS_PLACEHOLDER]" in user_prompt:
            predictive_results = perform_predictive_analysis(df, table_name, ge_summary)
            analysis_summary = format_analysis_summary(predictive_results)
            user_prompt = user_prompt.replace("[PREDICTIVE_ANALYSIS_PLACEHOLDER]", analysis_summary)        # Step 4: Call AI Foundry for insights
        response = call_ai_foundry_agent(user_prompt)
        
        return func.HttpResponse(response, status_code=200)    except Exception as e:
        logging.error(f"Function execution error: {str(e)}")
        return func.HttpResponse(f"Error: {str(e)}", status_code=500)

Step 4: Multi-Source Data Integration

Implement dynamic data source loading based on environment configuration:

def get_data_source():
    """Determine and load data from configured source"""
    data_source_type = os.getenv("DATA_SOURCE_TYPE", "unity_catalog").lower()
    
    if data_source_type == "unity_catalog":
        return load_unity_catalog_data()
    elif data_source_type == "web_api":
        return load_web_api_data()
    elif data_source_type == "web_scraping":
        return load_web_scraping_data()
    elif data_source_type == "sql_server":
        return load_sql_server_data()
    else:
        logging.warning(f"Unknown data source: {data_source_type}")
        return load_unity_catalog_data()  # Fallback

def load_unity_catalog_data():
    """Load data from Databricks Unity Catalog"""
    host = os.getenv("DATABRICKS_HOST")
    token = os.getenv("DATABRICKS_TOKEN")
    http_path = os.getenv("HTTP_PATH")
    catalog = os.getenv("DATABRICKS_CATALOG", "default_catalog")
    schema = os.getenv("DATABRICKS_SCHEMA", "default_schema")
    table = os.getenv("DATABRICKS_TABLE", "default_table")
    
    with sql.connect(
        server_hostname=host,
        http_path=http_path,
        access_token=token
    ) as connection:
        with connection.cursor() as cursor:
            cursor.execute(f"USE CATALOG `{catalog}`")
            cursor.execute(f"USE SCHEMA `{schema}`")
            
            limit = int(os.getenv("DATABRICKS_LIMIT", "1000"))
            cursor.execute(f"SELECT * FROM `{catalog}`.`{schema}`.`{table}` LIMIT {limit}")
            
            rows = cursor.fetchall()
            columns = [col[0] for col in cursor.description]
            df = pd.DataFrame(rows, columns=columns)
            
            return df, f"{catalog}.{schema}.{table}"

Step 5: Great Expectations Configuration

def run_great_expectations_validation():
    """Execute Great Expectations validation workflow"""
    # Initialize Great Expectations context
    context = gx.get_context()
    
    # Get data source and create validator
    df, table_name = get_data_source()
    validator = context.sources.pandas_default.read_dataframe(df)
    
    # Execute validations based on data source type
    data_source_type = os.getenv("DATA_SOURCE_TYPE", "unity_catalog")
    
    if data_source_type == "unity_catalog":
        return execute_unity_catalog_validations(validator, df, table_name)
    else:
        return execute_web_data_validations(validator, df, table_name)

def execute_unity_catalog_validations(validator, df, table_name):
    """Execute specific validations for Unity Catalog data"""
    results = []
    summary = f"Unity Catalog GE Validation Results for {table_name}:\\n\\n"
    
    # Table-level validations
    result1 = validator.expect_table_row_count_to_be_between(min_value=1, max_value=100000)
    results.append(result1)
    summary += f"- Table row count: {'PASSED' if result1.success else 'FAILED'}\\n"
    
    # Column-level validations for critical columns
    critical_columns = ["WellCode", "Process", "record_create_date"]
    available_columns = [col for col in critical_columns if col in df.columns]
    
    for col in available_columns:
        # Check column existence
        result = validator.expect_column_to_exist(column=col)
        results.append(result)
        summary += f"- Column '{col}' exists: {'PASSED' if result.success else 'FAILED'}\\n"
        
        # Check for null values
        result = validator.expect_column_values_to_not_be_null(column=col)
        results.append(result)
        null_count = result.result.get("unexpected_count", 0)
        total_count = result.result.get("element_count", len(df))
        null_percentage = (null_count / total_count * 100) if total_count > 0 else 0
        summary += f"- Column '{col}' non-null: {'PASSED' if result.success else 'FAILED'} ({null_percentage:.1f}% nulls)\\n"
    
    # Calculate success metrics
    success_count = sum(1 for r in results if r.success)
    success_rate = (success_count / len(results) * 100) if results else 0
    
    summary += f"\\nOverall Results:\\n"
    summary += f"- Success Rate: {success_rate:.1f}% ({success_count}/{len(results)})\\n"
    summary += f"- Data Source: Unity Catalog ({table_name})\\n"
    summary += f"- Data Rows: {len(df):,}\\n"
    summary += f"- Data Columns: {len(df.columns)}\\n"
    
    return summary

Step 6: Azure AI Foundry Integration

def call_ai_foundry_agent(prompt_text):
    """Integrate with Azure AI Foundry for intelligent insights"""
    try:
        # Primary: Azure AI Foundry with project-based agents
        if os.getenv("AZURE_AI_AGENT_ID") and os.getenv("AZURE_AI_PROJECT_ENDPOINT"):
            return call_ai_foundry_with_project(prompt_text)
        else:
            return call_openai_fallback(prompt_text)
    except Exception as e:
        logging.warning(f"AI Foundry error: {e}, falling back to OpenAI")
        return call_openai_fallback(prompt_text)

def call_ai_foundry_with_project(prompt_text):
    """Advanced AI Foundry integration with project agents"""
    agent_id = os.getenv("AZURE_AI_AGENT_ID")
    project_endpoint = os.getenv("AZURE_AI_PROJECT_ENDPOINT")
    
    # Initialize AI project client
    project = AIProjectClient(
        credential=DefaultAzureCredential(),
        endpoint=project_endpoint
    )
    
    # Get agent and create conversation thread
    agent = project.agents.get_agent(agent_id)
    thread = project.agents.threads.create()
    
    # Send message to agent
    message = project.agents.messages.create(
        thread_id=thread.id,
        role="user",
        content=prompt_text
    )
    
    # Process with AI agent
    run = project.agents.runs.create_and_process(
        thread_id=thread.id,
        agent_id=agent.id
    )
    
    # Extract response
    messages = project.agents.messages.list(
        thread_id=thread.id, 
        order=ListSortOrder.ASCENDING
    )
    
    for message in messages:
        if message.role == "assistant" and message.text_messages:
            return message.text_messages[-1].text.value
    
    return "No response from AI agent"

Step 7: Predictive Analytics Implementation (optional-work in progress)

def perform_predictive_analysis(df, table_name, ge_summary):
    """Perform comprehensive predictive analysis"""
    analysis_results = {
        "table_name": table_name,
        "analysis_timestamp": datetime.now().isoformat(),
        "data_shape": {"rows": len(df), "columns": len(df.columns)},
        "quality_scores": {},
        "anomaly_detection": {},
        "trend_analysis": {},
        "recommendations": []
    }
    

    quality_scores = calculate_quality_scores(df, ge_summary)
    analysis_results["quality_scores"] = quality_scores
    

    anomalies = detect_data_quality_anomalies(df)
    analysis_results["anomaly_detection"] = anomalies
    

    trends = analyze_temporal_trends(df)
    analysis_results["trend_analysis"] = trends
    
  
    recommendations = generate_recommendations(df, ge_summary, anomalies, trends)
    analysis_results["recommendations"] = recommendations
    
    return analysis_results

def calculate_quality_scores(df, ge_summary):
    """Calculate comprehensive data quality scores"""
    scores = {
        "overall_quality_score": 0.0,
        "completeness_score": 0.0,
        "consistency_score": 0.0,
        "uniqueness_score": 0.0,
        "validity_score": 0.0
    }
    
    # Completeness Score (null percentage)
    total_cells = df.shape[0] * df.shape[1]
    null_cells = df.isnull().sum().sum()
    scores["completeness_score"] = max(0, (1 - null_cells / total_cells) * 100)
    
    # Uniqueness Score (duplicate detection)
    duplicate_rows = df.duplicated().sum()
    scores["uniqueness_score"] = max(0, (1 - duplicate_rows / len(df)) * 100)
    
    # Validity Score (based on GE validation results)
    ge_success_count = ge_summary.count("PASSED")
    ge_total_count = ge_summary.count("PASSED") + ge_summary.count("FAILED")
    scores["validity_score"] = (ge_success_count / ge_total_count * 100) if ge_total_count > 0 else 0
    
    # Consistency Score (data type consistency)
    scores["consistency_score"] = calculate_consistency_score(df)
    
    # Overall Score (weighted average)
    scores["overall_quality_score"] = (
        scores["completeness_score"] * 0.3 +
        scores["consistency_score"] * 0.25 +
        scores["uniqueness_score"] * 0.25 +
        scores["validity_score"] * 0.2
    )
    
    return scores

Step 8: Deployment Configuration

Step 9: Function App Configuration Files

# host.json - Function App Configuration
{
    "version": "2.0",
    "functionTimeout": "00:10:00",
    "extensions": {
        "http": {
            "routePrefix": "api"
        }
    },
    "logging": {
        "applicationInsights": {
            "samplingSettings": {
                "isEnabled": true
            }
        }
    }
}

# function.json - Function Binding Configuration
{
    "scriptFile": "__init__.py",
    "bindings": [
        {
            "authLevel": "function",
            "type": "httpTrigger",
            "direction": "in",
            "name": "req",
            "methods": ["post"]
        },
        {
            "type": "http",
            "direction": "out",
            "name": "$return"
        }
    ]
}

Step 10: Testing and Validation

Testing Checklist:

Test local function execution with sample data
Validate Great Expectations rules with known good/bad data
Verify Azure AI Foundry integration and responses
Test all supported data source types
Perform load testing with larger datasets
Validate error handling and fallback mechanisms

# Sample test request
curl -X POST "https://your-function-app.azurewebsites.net/api/DatabricksConnector" \
  -H "Content-Type: application/json" \
  -H "x-functions-key: YOUR_FUNCTION_KEY" \
  -d '{
    "prompt": "Analyze the data quality for [DATAFRAME_PLACEHOLDER] using Great Expectations validation results: [GE_SUMMARY_PLACEHOLDER] and provide insights with predictive analysis: [PREDICTIVE_ANALYSIS_PLACEHOLDER]"
  }'

Azure Function Testing & Validation

Complete Testing Workflow:

Follow this comprehensive testing process to validate your Azure Function deployment and verify the Azure AI Foundry integration is working correctly.

Initialize Azure Function Host

Start the Azure Function host in your VS Code environment. The function will be running locally and ready to accept HTTP requests. Ensure all dependencies are installed and the function host starts without errors.

2. Configure Postman API Testing

Set up Postman to test your Azure Function endpoint with the following configuration:

Method: POST
URL: Your Azure Function endpoint (local: http://localhost:7071/api/DatabricksConnector or deployed URL)
Headers: Content-Type: application/json
Body: Raw JSON payload

{
  "prompt": "Provide a Great Expectations data quality summary: {{GE_SUMMARY}}"
}

You can see response from Azure Function

Expected Postman Response

Upon successful execution, we should receive a comprehensive response from the Azure Function containing AI-generated insights, data quality analysis, and recommendations based on your Great Expectations validation results.

Azure AI Foundry Integration Verification

Now lets validate to make sure it took some suggestions from Azure AI foundry.

Open Azure AI foundry → Go to agents → click on your agent.
Open My Threads → click on last latest thread and then you can see response whatever we have asked on Postman.

3. For comprehensive analysis and debugging extended threads, then click on Try in playground option → click on “Thread Logs”

Expected Output & Testing Objectives:

The Azure Function should return comprehensive AI-powered insights combining data previews, validation results, quality scores, anomaly detection, and intelligent recommendations for data quality improvement. This testing workflow is specifically designed to test and compare results between Databricks Genie and Azure AI Foundry, even though both platforms utilize similar AI agents on the backend.

Integration Verification:

The Azure AI Foundry console should display corresponding thread activity showing your agent successfully processing the data quality analysis request with full traceability.

Databricks Genie vs Azure AI Foundry Comparison

While both platforms use similar AI agents on the backend, there are key architectural and strategic differences that make Azure AI Foundry more suitable for this enterprise data quality validation use case.

References & Resources

https://learn.microsoft.com/en-us/azure/databricks/dlt/unity-catalog
https://learn.microsoft.com/en-us/azure/azure-functions/functions-reference-python?tabs=get-started%2Casgi%2Capplication-level&pivots=python-mode-decorators
https://docs.greatexpectations.io/docs/home/
https://learn.microsoft.com/en-us/azure/ai-foundry/
https://learn.microsoft.com/en-us/azure/machine-learning/?view=azureml-api-2