This is a part2 of Great Expectations with GitHub, as in earlier part we have seen how to start with Great Expectations usage with Databricks. Now let's check how we can leverage the same using GitHub to make it as Re-Usable component.
Imagine if your data quality checks ran automatically every time someone made changes to your code or data. No more manual testing, no more forgetting to run validations, and no more broken pipelines reaching production!
Press enter or click to view image in full size
Why Automate Data Quality?
Developer commits code GitHub Actions automatically triggers Great Expectations runs all checks finally Issues caught before production.
Complete CI/CD Architecture
when someone Pushes code to feature/dataquality branch GitHub Actions Automatically triggers validation workflow. It will connect to real Databricks data may be on Azure SQL Server or Unity Catalog. Followed by Running comprehensive quality checks based on Business Rules/ Table rules etc (Defined in Part1). It will run Deep data analysis with quality scores and generate Detailed JSON results and insights. Finally your new Rules/Conditions will marked as Pass/Fail Automatic approval or rejection.
The 7-Step Automation Process
Here are the steps to achieve Great expectations validations using Github actions. The workflow automatically starts when you push code to the feature/dataquality branch or manually trigger it.
Step 1: Trigger Setup
The workflow automatically starts when you push code to the feature/dataquality branch or manually trigger it.
Step 2: Environment Preparation
GitHub Actions sets up with Ubuntu environment with Python 3.12 and all required packages. Make sure to select latest version of Python packages.
- name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.12'
- name: Install system dependencies run: | sudo apt-get update sudo apt-get install -y build-essential tree
The system loads up to 1,000 records from Unity Catalog and runs comprehensive Great Expectations checks. only for testing I used 1000 records but in real time Ingested over 1 million records and took less than 2 hours to finish entire execution.
cursor.execute(f"SELECT * FROM `{catalog}`.`{schema}`.`{table}` LIMIT 1000")
Great Expectations performs deep data profiling, analyzing every column and generating quality scores.
try:
datasource_name = "profiling_datasource"
try:
print(f"Creating new Fluent datasource: {datasource_name}") datasource = context.data_sources.add_pandas(name=datasource_name) except Exception as ds_error:
print(f"Datasource creation failed, trying to get existing: {ds_error}") try: datasource = context.data_sources.get(datasource_name) print(f"Using existing datasource: {datasource_name}") except Exception as get_ds_error: print(f"Neither create nor get datasource worked: {get_ds_error}") raise Exception(f"Cannot create or get datasource '{datasource_name}': create={ds_error}, get={get_ds_error}")
data_asset_name = "unity_catalog_data" try:
print(f"Creating new asset: {data_asset_name}") data_asset = datasource.add_dataframe_asset(name=data_asset_name) except Exception as asset_error:
print(f"Asset creation failed, trying to get existing: {asset_error}") try: data_asset = datasource.get_asset(data_asset_name) print(f"Using existing asset: {data_asset_name}") except Exception as get_error: print(f"Neither create nor get worked: {get_error}") raise Exception(f"Cannot create or get asset '{data_asset_name}': create={asset_error}, get={get_error}")
batch_definition_name = "profiling_batch" try:
print(f"Creating new batch definition: {batch_definition_name}") batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name) except Exception as batch_def_error:
print(f"Batch definition creation failed, trying to get existing: {batch_def_error}") try: batch_definition = data_asset.get_batch_definition(batch_definition_name) print(f"Using existing batch definition: {batch_definition_name}") except Exception as get_batch_error: print(f"Neither create nor get batch definition worked: {get_batch_error}") raise Exception(f"Cannot create or get batch definition '{batch_definition_name}': create={batch_def_error}, get={get_batch_error}")
print("Getting batch with dataframe...") batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
print("Creating validator...") validator = context.get_validator(batch=batch) print("Great Expectations validator created successfully")
print(f"Creating new Fluent datasource: {datasource_name}") datasource = context.data_sources.add_pandas(name=datasource_name) except Exception as ds_error: # If creation fails, try to get existing print(f"Datasource creation failed, trying to get existing: {ds_error}") try: datasource = context.data_sources.get(datasource_name) print(f"Using existing datasource: {datasource_name}") except Exception as get_ds_error: print(f"Neither create nor get datasource worked: {get_ds_error}") raise Exception(f"Cannot create or get datasource '{datasource_name}': create={ds_error}, get={get_ds_error}")
# Create data asset from the dataframe data_asset_name = "unity_catalog_data" try: # Create new asset (don't try to get existing first) print(f"Creating new asset: {data_asset_name}") data_asset = datasource.add_dataframe_asset(name=data_asset_name) except Exception as asset_error: # If creation fails, try to get existing print(f"Asset creation failed, trying to get existing: {asset_error}") try: data_asset = datasource.get_asset(data_asset_name) print(f"Using existing asset: {data_asset_name}") except Exception as get_error: print(f"Neither create nor get worked: {get_error}") raise Exception(f"Cannot create or get asset '{data_asset_name}': create={asset_error}, get={get_error}")
# Create batch definition and get batch using Fluent API batch_definition_name = "profiling_batch" try: # Create new batch definition (don't try to get existing first) print(f"Creating new batch definition: {batch_definition_name}") batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name) except Exception as batch_def_error: # If creation fails, try to get existing print(f"Batch definition creation failed, trying to get existing: {batch_def_error}") try: batch_definition = data_asset.get_batch_definition(batch_definition_name) print(f"Using existing batch definition: {batch_definition_name}") except Exception as get_batch_error: print(f"Neither create nor get batch definition worked: {get_batch_error}") raise Exception(f"Cannot create or get batch definition '{batch_definition_name}': create={batch_def_error}, get={get_batch_error}")
# Get the actual batch with our dataframe print("Getting batch with dataframe...") batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
# Get validator for profiling print("Creating validator...") validator = context.get_validator(batch=batch) print("Great Expectations validator created successfully")
# Calculate overall metrics total_expectations = sum(len(col["gx_expectations"]) for col in profile_results["column_profiles"].values()) successful_expectations = sum( sum(1 for exp in col["gx_expectations"] if exp["success"]) for col in profile_results["column_profiles"].values() )
profile_results["table_summary"].update({ "total_gx_expectations": total_expectations, "successful_gx_expectations": successful_expectations, "gx_success_rate": round((successful_expectations / total_expectations * 100), 2) if total_expectations > 0 else 0, "duplicate_rows": int(df.duplicated().sum()), "duplicate_percentage": round(df.duplicated().sum() / len(df) * 100, 2), "overall_completeness": round(sum(col["completeness_score"] for col in profile_results["column_profiles"].values()) / len(profile_results["column_profiles"]), 2) })
# Save profiling results with open("gx_data_profiling_results.json", "w") as f: json.dump(profile_results, f, indent=2, default=str)
import traceback traceback.print_exc() # Do not return None - raise the exception to fail the workflow raise Exception(f"Great Expectations profiling failed (mandatory): {e}")
# Step 1: Perform Great Expectations data profiling (MANDATORY) profiling_results = perform_gx_data_profiling(context, df, table_name)
# ENFORCE: Great Expectations is mandatory - fail if profiling failed if profiling_results is None: raise Exception("CRITICAL: Great Expectations profiling failed. GX is mandatory for this workflow.")
# Step 2: Great Expectations validation using the SAME Fluent API approach results = []
print("Running Great Expectations validations using Fluent API...")
# Use the same Fluent Datasource approach as profiling (NO FALLBACK) try: # Get the same datasource used for profiling datasource_name = "profiling_datasource" datasource = context.data_sources.get(datasource_name)
# Get the same asset and batch definition created during profiling data_asset = datasource.get_asset("unity_catalog_data") batch_definition = data_asset.get_batch_definition("profiling_batch")
# Create a new batch with the same dataframe for validation batch = batch_definition.get_batch(batch_parameters={"dataframe": df}) validator = context.get_validator(batch=batch)
print("Great Expectations validator created using Fluent API")
# Run Great Expectations validations (MANDATORY) result1 = validator.expect_table_row_count_to_be_between(min_value=1, max_value=100000) results.append(result1) print(f"Table row count validation: {'PASSED' if result1.success else 'FAILED'}")
critical_columns = ["WellCode", "Process", "record_create_date"] for col in critical_columns: if col in df.columns: result = validator.expect_column_values_to_not_be_null(column=col) results.append(result) print(f"Column '{col}' null check: {'PASSED' if result.success else 'FAILED'}")
for col in critical_columns: if col in df.columns: result = validator.expect_column_to_exist(column=col) results.append(result) print(f"Column '{col}' exists check: {'PASSED' if result.success else 'FAILED'}")
print(f"All Great Expectations validations completed using Fluent API")
except Exception as gx_error: print(f"CRITICAL ERROR: Great Expectations validation failed: {gx_error}") import traceback traceback.print_exc() raise Exception(f"Great Expectations is mandatory but failed: {gx_error}")
success_count = sum(1 for r in results if r.get("success", False)) total_count = len(results) success_rate = float(success_count / total_count * 100) if total_count > 0 else 0.0 overall_success = all(r.get("success", False) for r in results)
if ps.get('column_quality_scores'): print(f\"\\nCOLUMN QUALITY SCORES (GX-based)\") for col, score in ps['column_quality_scores'].items(): status = 'GOOD' if score >= 80 else 'MEDIUM' if score >= 60 else 'POOR' print(f\" {col}: {score}% ({status})\") " else echo "No validation report found" fi
echo "" if [ -f "gx_data_profiling_results.json" ]; then
python -c " import json with open('gx_data_profiling_results.json', 'r') as f: profile = json.load(f)
insights = profile.get('data_quality_insights', []) if insights: print('Quality Issues Found (GX-based):') for insight in insights[:5]: # Show first 5 issues severity_icon = 'ERROR' if insight['severity'] == 'error' else 'WARNING' if insight['severity'] == 'warning' else 'INFO' gx_marker = ' [GX]' if insight.get('gx_based') else '' print(f\" {severity_icon}: {insight['type']}: {insight['description']}{gx_marker}\") if len(insights) > 5: print(f\" ... and {len(insights) - 5} more issues (see GX profiling artifact)\") else: print('No significant quality issues detected by Great Expectations')
# Show some column details columns = profile.get('column_profiles', {}) if columns: print(f\"\\nColumn GX Expectations Summary:\") for col_name, col_data in list(columns.items())[:3]: # Show first 3 columns gx_expectations = col_data.get('gx_expectations', []) successful = sum(1 for exp in gx_expectations if exp['success']) total = len(gx_expectations) print(f\" {col_name}: {successful}/{total} GX expectations passed\") if len(columns) > 3: print(f\" ... and {len(columns) - 3} more columns (see full artifact)\") " else echo "No GX profiling results found" fi
