🧠 Data Quality Fundamentals

Q: What is Great Expectations?

Great Expectations (GX) is an open-source Python framework for data quality testing. It lets you define expectations about your data, validate against them, generate documentation, and integrate with pipelines to catch data issues before they reach production.

Q: What are the 6 dimensions of data quality?

The six dimensions are: accuracy, completeness, consistency, timeliness, validity, and uniqueness. Great Expectations provides expectations for each dimension.

Q: What is an expectation in Great Expectations?

An expectation is a declarative statement about your data, such as 'column X should never be null' or 'column Y should have values between 0 and 100'. Great Expectations has 300+ built-in expectations.

Q: How does Great Expectations differ from dbt tests?

dbt tests are SQL-based and run inside your warehouse, focused on transformations. Great Expectations is Python-based, works with any data source (dataframes, SQL, Spark), and offers richer expectations, data docs, and profiling.

Q: What is a data contract?

A data contract is a formal agreement between data producers and consumers that specifies schema, semantics, quality SLAs, and versioning. Great Expectations can enforce data contracts by validating data against defined expectations.

🎯 What is Data Quality?

Data Quality is the measure of how fit data is for its intended purpose. Poor data quality costs organizations an average of $12.9 million per year (Gartner). Great Expectations (GX) is the leading open-source framework for testing and monitoring data quality in Python pipelines.

Why Data Quality Matters

Trust: Reliable decisions. Cost: $12.9M/year lost. AI/ML: Garbage in → garbage out. Compliance: Regulatory. Trust: Stakeholder confidence. Speed: Fewer incidents.

📌 Fit for Purpose📌 Trust📌 Testing📌 Monitoring

# Data Quality & Great Expectations
# The framework for trustworthy data.

# CONCEPTUAL OVERVIEW
print("""
=== Data Quality ===

What it is:
  - Fitness for purpose
  - Accuracy, completeness, etc.
  - Trust in data
  - Measurable

Why it matters:
  - Bad data → bad decisions
  - $12.9M/year average loss
  - AI/ML needs clean data
  - Compliance requires it

=== Great Expectations ===

What it is:
  - Python framework
  - Data quality testing
  - Open source (Apache 2.0)
  - Declarative expectations
  - Data docs
  - Validation

Key concepts:
  - Expectation: Assertion about data
  - Suite: Collection of expectations
  - Checkpoint: Validation run
  - Data Docs: HTML report
  - Profiler: Auto-generate expectations
""")

# WHY IT MATTERS
print("=== Cost of Bad Data ===")
print("""
Gartner estimates:
  - $12.9M/year average cost
  - 60% of data projects fail
  - 1-10-100 rule:
    - $1 to prevent
    - $10 to correct
    - $100 to fix downstream

Real-world:
  - Wrong decisions
  - Lost revenue
  - Compliance fines
  - Reputation damage
  - Wasted time
""")

# REAL-WORLD ADOPTION
print("=== Who Uses Great Expectations ===")
print("""
Companies:
  - Google
  - Adobe
  - Instacart
  - Heineken
  - AstraZeneca
  - Roche

Use cases:
  - ETL/ELT pipelines
  - ML feature validation
  - Data warehouses
  - Streaming
  - Data contracts
""")

📊 6 Dimensions of Data Quality

🎯 The Six Dimensions

Data quality is multidimensional. The six core dimensions are accuracy, completeness, consistency, timeliness, validity, and uniqueness.

The 6 Dimensions

Accuracy: Correct values. Completeness: No missing. Consistency: No contradictions. Timeliness: Up-to-date. Validity: Conforms. Uniqueness: No dups.

📊 Accuracy📊 Completeness📊 Consistency📊 Timeliness

# 6 Dimensions of Data Quality
print("=== 1. ACCURACY ===")
print("""
Accuracy:
  - Data correctly represents reality
  - Example: Customer age is correct
  - Issues: Typos, errors, outdated

Expectations:
  - expect_column_values_to_be_between
  - expect_column_values_to_match_regex
  - expect_column_values_to_be_in_set

Example:
  email column matches email regex
  age column between 0 and 150
""")

print("\n=== 2. COMPLETENESS ===")
print("""
Completeness:
  - No missing values
  - All required fields present
  - Example: All users have emails

Expectations:
  - expect_column_values_to_not_be_null
  - expect_table_row_count_to_be_between
  - expect_column_values_to_be_unique

Example:
  user_id never null
  at least 1000 rows
""")

print("\n=== 3. CONSISTENCY ===")
print("""
Consistency:
  - No contradictions
  - Same data across systems
  - Example: Customer ID matches everywhere

Expectations:
  - expect_column_pair_values_to_be_equal
  - expect_multicolumn_values_to_be_unique
  - expect_column_values_to_be_in_set

Example:
  order.total = sum(line_items.amount)
  customer.id matches across tables
""")

print("\n=== 4. TIMELINESS ===")
print("""
Timeliness:
  - Up-to-date
  - Fresh data
  - Example: Data refreshed daily

Expectations:
  - expect_column_max_to_be_between
  - expect_table_row_count_to_be_between
  - expect_column_values_to_be_dateutil_parseable

Example:
  latest timestamp within 24 hours
  daily row count > 1M
""")

print("\n=== 5. VALIDITY ===")
print("""
Validity:
  - Conforms to rules
  - Domain constraints
  - Example: Age is positive

Expectations:
  - expect_column_values_to_be_in_set
  - expect_column_values_to_match_regex
  - expect_column_values_to_be_between

Example:
  status in ['active', 'inactive']
  phone matches format
  age > 0
""")

print("\n=== 6. UNIQUENESS ===")
print("""
Uniqueness:
  - No duplicates
  - One record per entity
  - Example: One customer record per ID

Expectations:
  - expect_column_values_to_be_unique
  - expect_compound_columns_to_be_unique
  - expect_column_proportion_of_unique_values_to_be_between

Example:
  user_id unique
  (email, date) compound unique
""")

print("\n=== Summary Table ===")
print(f"{'Dimension':<20} {'Question':<30} {'Example':<25}")
print("-" * 75)
print(f"{'Accuracy':<20} {'Is it correct?':<30} {'Email valid':<25}")
print(f"{'Completeness':<20} {'Is it all there?':<30} {'No nulls':<25}")
print(f"{'Consistency':<20} {'Does it agree?':<30} {'IDs match':<25}")
print(f"{'Timeliness':<20} {'Is it current?':<30} {'Fresh <24h':<25}")
print(f"{'Validity':<20} {'Does it conform?':<30} {'Age > 0':<25}")
print(f"{'Uniqueness':<20} {'Is it unique?':<30} {'No dups':<25}")

🎯 Great Expectations

🎯 The Data Quality Framework

Great Expectations (GX) is an open-source Python framework for data quality. You define expectations (assertions) about your data, validate against them, and generate beautiful Data Docs.

Great Expectations

Python: Framework. Declarative: Assertions. 300+ Expectations: Built-in. Data Docs: HTML reports. Pipelines: Integration. Open: Apache 2.0.

🎯 Python🎯 Expectations🎯 Data Docs🎯 Open Source

# Great Expectations — deep dive
print("=== What is Great Expectations? ===")
print("""
Great Expectations:
  - Python framework
  - Data quality testing
  - Declarative
  - Open source (Apache 2.0)
  - 300+ expectations
  - Data docs
  - Pipeline integration

Created by:
  - Superconductive (now GX)
  - 2017

Core concepts:
  - Expectation: Assertion about data
  - Expectation Suite: Collection
  - Checkpoint: Validation run
  - Data Docs: HTML report
  - Profiler: Auto-generate
  - Data Context: Project config
  - Datasource: Data connection
  - Batch: Sample of data
""")

print("\n=== Installation ===")
print("""
# Install
pip install great_expectations

# Initialize project
great_expectations init

# Project structure:
great_expectations/
  ├── great_expectations.yml
  ├── expectations/
  ├── checkpoints/
  ├── data_docs/
  ├── plugins/
  └── uncommitted/
""")

print("\n=== Quick Start ===")
print("""
import great_expectations as gx

# Get context
context = gx.get_context()

# Connect to data
datasource = context.sources.add_pandas("my_datasource")
asset = datasource.add_csv_asset(
    name="users",
    filepath_or_buffer="users.csv"
)

# Create batch
batch_request = asset.build_batch_request()
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="users_suite"
)

# Define expectations
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_unique("user_id")
validator.expect_column_values_to_be_between("age", 0, 150)

# Save
validator.save_expectation_suite()

# Validate
checkpoint = context.add_or_update_checkpoint(
    name="my_checkpoint",
    validator=validator,
)
result = checkpoint.run()
""")

print("\n=== Key Features ===")
print("""
1. DECLARATIVE
   - Assertions, not code
   - Readable
   - Self-documenting

2. 300+ EXPECTATIONS
   - Column
   - Table
   - Multicolumn
   - Distributional

3. DATA DOCS
   - HTML reports
   - Shareable
   - Versioned

4. CHECKPOINTS
   - Batch validation
   - Actions
   - Alerts

5. PROFILER
   - Auto-generate
   - Data exploration
   - Onboarding

6. INTEGRATION
   - Airflow
   - dbt
   - Spark
   - Pandas
   - SQL
""")

📋 Expectations

🎯 The Building Blocks

Expectations are declarative assertions about data. GX has 300+ built-in expectations covering all data quality dimensions.

Expectations

Column: Per-column. Table: Whole table. Multicolumn: Across columns. Distributional: Stats. Custom: User-defined. 300+: Built-in.

📋 Column📋 Table📋 Multicolumn📋 Custom

# Expectations
print("=== Common Expectations ===")
print("""
# Column expectations
validator.expect_column_to_exist("email")
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_unique("email")
validator.expect_column_values_to_be_in_set("status", ["active", "inactive"])
validator.expect_column_values_to_be_between("age", 0, 150)
validator.expect_column_values_to_match_regex("email", r"^[^@]+@[^@]+\\.[^@]+$")
validator.expect_column_values_to_be_increasing("timestamp")
validator.expect_column_mean_to_be_between("salary", 30000, 200000)
validator.expect_column_stdev_to_be_between("salary", 1000, 50000)
validator.expect_column_values_to_be_of_type("age", "int")

# Table expectations
validator.expect_table_row_count_to_be_between(1000, 1000000)
validator.expect_table_column_count_to_equal(10)
validator.expect_table_columns_to_match_ordered_list(["id", "name", "age"])

# Multicolumn
validator.expect_compound_columns_to_be_unique(["user_id", "date"])
validator.expect_column_pair_values_a_to_be_greater_than_b("end_date", "start_date")
validator.expect_select_column_values_to_be_unique_within_record(["email", "phone"])
""")

print("\n=== Expectation Suite ===")
print("""
# Create suite
suite = context.create_expectation_suite("users_suite")

# Add expectations
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite=suite
)
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_unique("user_id")
validator.save_expectation_suite()

# Suite is stored as JSON:
{
  "expectation_suite_name": "users_suite",
  "expectations": [
    {
      "expectation_type": "expect_column_values_to_not_be_null",
      "kwargs": {"column": "user_id"},
      "meta": {}
    },
    {
      "expectation_type": "expect_column_values_to_be_unique",
      "kwargs": {"column": "user_id"},
      "meta": {}
    }
  ]
}
""")

print("\n=== Custom Expectations ===")
print("""
from great_expectations.expectations.expectation import ColumnExpectation

class ExpectColumnValuesToBeEven(ColumnExpectation):
    \"\"\"Expect column values to be even.\"\"\"
    
    def _validate(self, configuration, metrics, runtime_configuration, execution_engine):
        # Custom validation logic
        return {"success": True, "result": {}}

# Register
context.custom_expectations["expect_column_values_to_be_even"] = \\
    ExpectColumnValuesToBeEven
""")

print("\n=== Expectation Categories ===")
print(f"{'Category':<25} {'Examples':<45}")
print("-" * 75)
print(f"{'Column existence':<25} {'expect_column_to_exist':<45}")
print(f"{'Null/unique':<25} {'not_null, unique':<45}")
print(f"{'Values':<25} {'in_set, between, regex':<45}")
print(f"{'Statistics':<25} {'mean, stdev, median':<45}")
print(f"{'Table':<25} {'row_count, column_count':<45}")
print(f"{'Multicolumn':<25} {'compound_unique, pair':<45}")

✔️ Validation

🎯 Running Expectations

Validation runs your expectations against real data and reports pass/fail. Results include detailed statistics, unexpected values, and partial success.

Validation

Run: Expectations. Results: Pass/fail. Stats: Details. Unexpected: Failed values. Partial: Mostly pass. Data Docs: Report.

✔️ Run✔️ Results✔️ Stats✔️ Reports

# Validation
print("=== Running Validation ===")
print("""
# Simple validation
results = validator.validate()

# Results structure
{
    "success": True,
    "results": [
        {
            "success": True,
            "expectation_config": {
                "expectation_type": "expect_column_values_to_not_be_null",
                "kwargs": {"column": "user_id"}
            },
            "result": {
                "element_count": 1000,
                "unexpected_count": 0,
                "unexpected_percent": 0.0,
                "partial_unexpected_list": []
            }
        }
    ],
    "meta": {...},
    "statistics": {
        "evaluated_expectations": 10,
        "successful_expectations": 10,
        "unsuccessful_expectations": 0,
        "success_percent": 100.0
    }
}
""")

print("\n=== Partial Success ===")
print("""
# Most expectations pass → success
# Unless mostly=0.9

validator.expect_column_values_to_be_in_set(
    "status",
    ["active", "inactive"],
    mostly=0.95  # 95% must pass
)

# Results:
{
    "success": True,  # >= 95% pass
    "result": {
        "unexpected_count": 30,
        "unexpected_percent": 3.0,  # 3% fail
        "partial_unexpected_list": ["pending", "cancelled", ...]
    }
}
""")

print("\n=== Data Docs Output ===")
print("""
# Generate Data Docs
context.build_data_docs()

# Open in browser
context.open_data_docs()

# Docs include:
  - Index (all validations)
  - Validation results (per run)
  - Expectation suite
  - Statistics
  - Unexpected values
""")

print("\n=== Programmatic Checks ===")
print("""
# Check results in code
results = checkpoint.run()

if results["success"]:
    print("✓ All validations passed")
    # Proceed with pipeline
else:
    print("✗ Validation failed")
    # Alert, fail, or quarantine
    for result in results["results"]:
        if not result["success"]:
            print(f"Failed: {result['expectation_config']['expectation_type']}")
            print(f"  Column: {result['expectation_config']['kwargs'].get('column')}")
            print(f"  Unexpected: {result['result'].get('unexpected_count')}")
    
    # Raise exception or alert
    raise ValueError("Data quality check failed")
""")

print("\n=== Exit Codes ===")
print("""
For CI/CD pipelines:

great_expectations checkpoint run my_checkpoint

Exit codes:
  0: All passed
  1: Some failed
  2: Error running
""")

📄 Data Docs

🎯 Beautiful HTML Reports

Data Docs are auto-generated HTML reports that document your expectations, validation results, and data statistics. They're shareable with stakeholders.

Data Docs

HTML: Static site. Shareable: S3, GCS. Index: All suites. Suite: Expectations. Results: Validations. Lineage: History.

📄 HTML📄 Shareable📄 Versioned📄 Lineage

# Data Docs
print("=== What are Data Docs? ===")
print("""
Data Docs:
  - Auto-generated HTML
  - Static site
  - Shareable
  - Versioned
  - Searchable

Includes:
  - Index of suites
  - Expectation suites
  - Validation results
  - Statistics
  - Unexpected values
  - Run history
""")

print("\n=== Generate ===")
print("""
# Build Data Docs
context.build_data_docs()

# Open in browser
context.open_data_docs()

# Specific resource
context.open_data_docs(resource_identifier={
    "type": "validation_result",
    "site_names": ["local_site"],
})
""")

print("\n=== Configuration ===")
print("""
# great_expectations.yml
data_docs_sites:
  local_site:
    class_name: SiteBuilder
    show_how_to_buttons: false
    store_backend:
      class_name: TupleFilesystemStoreBackend
      base_directory: uncommitted/data_docs/local_site/
    site_index_builder:
      class_name: DefaultSiteIndexBuilder
  
  s3_site:
    class_name: SiteBuilder
    store_backend:
      class_name: TupleS3StoreBackend
      bucket: my-gx-docs
      prefix: data_docs/
""")

print("\n=== Hosting ===")
print("""
Host Data Docs:
  - S3 + CloudFront
  - GCS + Firebase
  - GitHub Pages
  - Netlify
  - Any static host

Share with:
  - Data team
  - Stakeholders
  - Compliance
  - Leadership

Benefits:
  - Transparency
  - Trust
  - Documentation
  - Onboarding
""")

print("\n=== Example Report ===")
print("""
# Data Docs Index
  - All suites
  - All validations
  - Search
  - Statistics

# Suite page
  - Expectation list
  - Descriptions
  - Examples
  - Code

# Validation page
  - Success/fail
  - Statistics
  - Unexpected values
  - Charts
  - History
""")

🎯 Checkpoints

🎯 Batch Validation

Checkpoints are the primary way to run validations in production. They bundle data, expectations, and actions (like alerts) into a single unit.

Checkpoints

Bundle: Data + suite. Actions: After validation. Schedules: Automated. Alerts: On failure. History: Tracked. Production: Standard.

🎯 Bundle🎯 Actions🎯 Alerts🎯 Production

# Checkpoints
print("=== What is a Checkpoint? ===")
print("""
Checkpoint:
  - Bundles validation
  - Data + Suite
  - Actions
  - Schedules
  - Production

Components:
  - Batch request
  - Expectation suite
  - Action list
  - Validator
""")

print("\n=== Create Checkpoint ===")
print("""
checkpoint = context.add_or_update_checkpoint(
    name="daily_validation",
    validations=[
        {
            "batch_request": batch_request,
            "expectation_suite_name": "users_suite",
        },
    ],
    action_list=[
        {
            "name": "store_validation_result",
            "action": {
                "class_name": "StoreValidationResultAction",
            },
        },
        {
            "name": "update_data_docs",
            "action": {
                "class_name": "UpdateDataDocsAction",
            },
        },
        {
            "name": "send_slack_notification",
            "action": {
                "class_name": "SlackNotificationAction",
                "slack_webhook": "${SLACK_WEBHOOK}",
                "notify_on": "failure",
            },
        },
    ],
)
""")

print("\n=== Run Checkpoint ===")
print("""
# Run
results = checkpoint.run()

# Check
if results["success"]:
    print("✓ Validation passed")
else:
    print("✗ Validation failed")
    # Alert

# Actions run automatically:
#   - Store results
#   - Update docs
#   - Send alerts
""")

print("\n=== Actions ===")
print("""
Common actions:
  - StoreValidationResultAction
  - StoreEvaluationParametersAction
  - StoreMetricsAction
  - UpdateDataDocsAction
  - SlackNotificationAction
  - EmailAction
  - PagerDutyAlertAction
  - MicrosoftTeamsNotificationAction
  - OpsgenieAlertAction
  - SNSNotificationAction
  - Custom actions

Actions run on:
  - Success
  - Failure
  - Both
""")

print("\n=== Orchestration ===")
print("""
# Airflow
from airflow.operators.python import PythonOperator
from great_expectations_provider.operators.great_expectations import (
    GreatExpectationsOperator
)

gx_task = GreatExpectationsOperator(
    task_id="validate_data",
    checkpoint_name="daily_validation",
    data_context_root_dir="/path/to/gx",
    fail_task_on_validation_failure=True,
)

# dbt
# Use dbt + GX via dbt-expectations or
# dbt + GX checkpoint after dbt run

# Dagster
from dagster import asset, AssetExecutionContext
from dagster_ge import ge_validation_op_factory

validate_users = ge_validation_op_factory(
    name="validate_users",
    datasource_name="my_datasource",
    suite_name="users_suite",
)
""")

print("\n=== Scheduling ===")
print("""
Checkpoints can be scheduled:
  - Cron in Airflow
  - Dagster schedules
  - Prefect schedules
  - Kubernetes CronJobs
  - Serverless (Lambda, Cloud Functions)

Common cadence:
  - After ETL
  - Hourly/daily
  - On data arrival
  - Before ML training
""")

🔍 Profiling

🎯 Auto-Generate Expectations

Profiling analyzes your data and automatically suggests expectations. Perfect for onboarding new datasets and discovering patterns.

Profiling

Analyze: Data statistics. Suggest: Expectations. Explore: Patterns. Onboard: New datasets. Review: Edit suggestions. Automate: Data discovery.

🔍 Analyze🔍 Suggest🔍 Explore🔍 Onboard

# Profiling
print("=== What is Profiling? ===")
print("""
Profiling:
  - Auto-generate expectations
  - Analyze data
  - Suggest patterns
  - Onboarding
  - Exploration

Types:
  - Onboarding: For new data
  - User-configurable: Custom
  - Rule-based: From rules
""")

print("\n=== Onboarding Profiler ===")
print("""
from great_expectations.profile.user_configurable_profiler import (
    UserConfigurableProfiler
)

# Profile
profiler = UserConfigurableProfiler(
    profile_dataset=validator,
    excluded_expectations=[
        "expect_column_values_to_be_unique",
        "expect_column_values_to_match_regex",
    ],
    ignored_columns=["id"],
    value_set_threshold="few",
)

suite = profiler.build_suite()

# Suite contains suggestions:
#   - expect_column_to_exist
#   - expect_column_values_to_not_be_null
#   - expect_column_values_to_be_in_set
#   - expect_column_values_to_be_between
#   - expect_column_values_to_be_of_type
#   - etc.
""")

print("\n=== Rule-Based Profiler ===")
print("""
from great_expectations.rule_based_profiler import RuleBasedProfiler

# Custom rules
profiler = RuleBasedProfiler(
    name="my_profiler",
    config_version=1.0,
    rules={
        "my_rule": {
            "domain_builder": {...},
            "parameter_builders": [...],
            "expectation_configuration_builders": [...],
        },
    },
    data_context=context,
)

suite = profiler.run().expectation_suite
""")

print("\n=== Workflow ===")
print("""
1. PROFILE
   - Analyze dataset
   - Generate suite
   
2. REVIEW
   - Inspect expectations
   - Remove false positives
   - Add custom
   
3. EDIT
   - Adjust thresholds
   - Add `mostly` parameter
   - Refine
   
4. SAVE
   - Save suite
   - Version control
   
5. VALIDATE
   - Run checkpoints
   - Monitor
   
6. ITERATE
   - Refine over time
""")

print("\n=== Benefits ===")
print("""
1. FAST ONBOARDING
   - No manual writing
   - Quick start
   
2. DISCOVER
   - Unknown patterns
   - Data quirks
   
3. BASELINE
   - Starting point
   - Refine later
   
4. EDUCATION
   - Learn data
   - Share knowledge
""")

📜 Data Contracts

🎯 Formal Data Agreements

A data contract is a formal agreement between data producers and consumers that specifies schema, semantics, quality SLAs, and versioning. GX enforces contracts by validating data against expectations.

Data Contracts

Schema: Structure. Semantics: Meaning. Quality: SLAs. Versioning: Compat. Ownership: Who. Enforcement: GX.

📜 Schema📜 Semantics📜 Quality📜 Versioning

# Data Contracts
print("=== What is a Data Contract? ===")
print("""
Data Contract:
  - Agreement between producer and consumer
  - Specifies schema
  - Semantics
  - Quality SLAs
  - Versioning
  - Ownership

Why:
  - Clarity
  - Trust
  - Compatibility
  - Accountability
  - Prevention
""")

print("\n=== Contract Structure ===")
print("""
# data_contract.yaml
apiVersion: v3.0.0
kind: DataContract
metadata:
  name: users
  version: 2.1.0
  owner: data-team@company.com
  description: User master data

schema:
  type: object
  properties:
    user_id:
      type: integer
      required: true
      unique: true
    email:
      type: string
      required: true
      format: email
    age:
      type: integer
      required: false
      minimum: 0
      maximum: 150

quality:
  freshness:
    max_age: 24h
  completeness:
    min_percent: 99.5
  accuracy:
    email_regex: true

sla:
  availability: 99.9%
  latency: <1h

versioning:
  breaking_changes: major
  additions: minor
  fixes: patch
""")

print("\n=== Enforcing with GX ===")
print("""
# Load contract
import yaml

with open("data_contract.yaml") as f:
    contract = yaml.safe_load(f)

# Create suite from contract
suite = context.create_expectation_suite("users_contract")

validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite=suite
)

# Enforce schema
for col, spec in contract["schema"]["properties"].items():
    validator.expect_column_to_exist(col)
    
    if spec.get("required"):
        validator.expect_column_values_to_not_be_null(col)
    
    if spec.get("unique"):
        validator.expect_column_values_to_be_unique(col)
    
    if spec.get("minimum") is not None:
        validator.expect_column_values_to_be_between(
            col,
            min_value=spec["minimum"],
            max_value=spec.get("maximum"),
        )

# Quality
if contract["quality"]["completeness"]["min_percent"]:
    validator.expect_column_values_to_not_be_null(
        "email",
        mostly=contract["quality"]["completeness"]["min_percent"] / 100,
    )

validator.save_expectation_suite()
""")

print("\n=== Benefits ===")
print("""
1. CLARITY
   - Explicit expectations
   - No ambiguity
   - Documented

2. TRUST
   - Producers accountable
   - Consumers confident

3. COMPATIBILITY
   - Versioning
   - Breaking changes known
   - Migration path

4. PREVENTION
   - Catch issues early
   - Before downstream
   - Save cost

5. ACCOUNTABILITY
   - Clear ownership
   - SLAs
   - Escalation
""")

print("\n=== Tools ===")
print("""
- Great Expectations (GX)
- Data Contract CLI
- Schemata
- Soda
- Custom (YAML + code)

Standards:
  - Open Data Contract Standard (ODCS)
  - Data Contract Specification
""")

🔌 Integration

🎯 GX in Your Stack

Great Expectations integrates with every major data tool: Airflow, dbt, Spark, Pandas, and more.

Integration

Airflow: Operator. dbt: Post-run. Spark: Native. Pandas: Native. SQL: SQLAlchemy. Cloud: S3, GCS, Azure.

🔌 Airflow🔌 dbt🔌 Spark🔌 SQL

# Integration
print("=== Airflow ===")
print("""
from great_expectations_provider.operators.great_expectations import (
    GreatExpectationsOperator
)

validate = GreatExpectationsOperator(
    task_id="validate_users",
    checkpoint_name="users_checkpoint",
    data_context_root_dir="/opt/airflow/gx",
    fail_task_on_validation_failure=True,
)

etl >> validate >> downstream
""")

print("\n=== dbt ===")
print("""
# Post-dbt validation
from airflow.operators.bash import BashOperator

dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command="dbt run",
)

gx_validate = BashOperator(
    task_id="gx_validate",
    bash_command="great_expectations checkpoint run dbt_checkpoint",
)

dbt_run >> gx_validate
""")

print("\n=== Spark ===")
print("""
from great_expectations.datasource.fluent import SparkDatasource

datasource = context.sources.add_spark("my_spark")
asset = datasource.add_dataframe_asset("users_df")

batch_request = asset.build_batch_request(
    dataframe=spark_df,
)

validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="users_suite",
)
""")

print("\n=== Pandas ===")
print("""
from great_expectations.datasource.fluent import PandasDatasource

datasource = context.sources.add_pandas("my_pandas")
asset = datasource.add_dataframe_asset("users_df")

batch_request = asset.build_batch_request(dataframe=df)
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="users_suite",
)
""")

print("\n=== SQL (SQLAlchemy) ===")
print("""
from great_expectations.datasource.fluent import SQLDatasource

datasource = context.sources.add_sql(
    name="my_postgres",
    connection_string="postgresql://user:pass@localhost:5432/mydb",
)

asset = datasource.add_table_asset(
    name="users",
    table_name="users",
)

batch_request = asset.build_batch_request()
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="users_suite",
)
""")

print("\n=== Cloud Storage ===")
print("""
# S3
datasource = context.sources.add_pandas_filesystem(
    name="s3_datasource",
    base_directory="s3://my-bucket/data/",
)

# GCS
datasource = context.sources.add_pandas_filesystem(
    name="gcs_datasource",
    base_directory="gs://my-bucket/data/",
)

# Azure
datasource = context.sources.add_pandas_filesystem(
    name="azure_datasource",
    base_directory="wasbs://container@account.blob.core.windows.net/data/",
)
""")

print("\n=== Databricks ===")
print("""
from great_expectations.datasource.fluent import DatabricksDatasource

datasource = context.sources.add_databricks(
    name="my_databricks",
    connection_string="databricks://...",
)

asset = datasource.add_table_asset(
    name="users",
    table_name="catalog.schema.users",
)

# Or via dbt + GX
""")

print("\n=== Orchestration Tools ===")
print(f"{'Tool':<20} {'Integration':<40}")
print("-" * 60)
print(f"{'Airflow':<20} {'great_expectations_provider':<40}")
print(f"{'Dagster':<20} {'dagster-ge':<40}")
print(f"{'Prefect':<20} {'prefect-great-expectations':<40}")
print(f"{'dbt':<20} {'dbt-expectations / post-hook':<40}")
print(f"{'Kedro':<20} {'kedro-great-expectations':<40}")

⚖️ vs Alternatives

🎯 GX vs Other Tools

Great Expectations competes with dbt tests, Soda, Monte Carlo, and others. Each has strengths.

Alternatives

dbt tests: SQL-based. Soda: SQL-based. Monte Carlo: Observability. Deequ: Spark. Pandera: Pandas. Custom: DIY.

⚖️ dbt tests⚖️ Soda⚖️ Monte Carlo⚖️ Deequ

Tool	Language	Best For
Great Expectations	Python	Multi-source, rich expectations
dbt tests	SQL	Transformations in dbt
Soda	SQL/YAML	SQL-first data quality
Monte Carlo	SaaS	Data observability
Deequ	Scala	Spark/EMR
Pandera	Python	Pandas dataframes
Elementary	SQL	dbt-native observability
Datafold	SaaS	Data diff + quality

# GX vs dbt Tests
print("=== GX vs dbt Tests ===")
print(f"{'Aspect':<25} {'Great Expectations':<25} {'dbt tests':<25}")
print("-" * 75)
print(f"{'Language':<25} {'Python':<25} {'SQL':<25}")
print(f"{'Data sources':<25} {'Any':<25} {'Warehouse':<25}")
print(f"{'Expectations':<25} {'300+':<25} {'Built-in + custom':<25}")
print(f"{'Data Docs':<25} {'Rich HTML':<25} {'Basic':<25}")
print(f"{'Profiling':<25} {'Yes':<25} {'No':<25}")
print(f"{'Learning curve':<25} {'Medium':<25} {'Low':<25}")
print(f"{'Best for':<25} {'Multi-source':<25} {'dbt-only':<25}")

print("\n=== GX vs Soda ===")
print(f"{'Aspect':<25} {'Great Expectations':<25} {'Soda':<25}")
print("-" * 75)
print(f"{'Language':<25} {'Python':<25} {'YAML/SQL':<25}")
print(f"{'Open source':<25} {'Yes':<25} {'Yes':<25}")
print(f"{'Expectations':<25} {'300+':<25} {'Basic':<25}")
print(f"{'Data Docs':<25} {'Rich':<25} {'Basic':<25}")
print(f"{'Profiling':<25} {'Yes':<25} {'No':<25}")
print(f"{'Best for':<25} {'Data teams':<25} {'Analysts':<25}")

print("\n=== When to Use Which ===")
print("""
USE GREAT EXPECTATIONS:
  ✓ Multi-source data
  ✓ Python pipelines
  ✓ Rich expectations
  ✓ Data docs needed
  ✓ Profiling

USE dbt TESTS:
  ✓ All data in dbt
  ✓ SQL-first
  ✓ Simple tests
  ✓ Warehouse-only

USE SODA:
  ✓ SQL-first
  ✓ YAML config
  ✓ Simple setup

USE MONTE CARLO:
  ✓ Managed observability
  ✓ ML-powered
  ✓ Budget

USE DEEQU:
  ✓ Spark/EMR
  ✓ Scala

USE PANDERA:
  ✓ Pandas only
  ✓ Type-safe
""")

✅ Best Practices

🎯 Doing Data Quality Right

Follow these best practices for effective data quality with Great Expectations.

Best Practices

Start small: Critical columns. Test early: In pipelines. Monitor: Continuously. Alert: On failure. Version: Suites. Document: Share.

✅ Start Small✅ Test Early✅ Monitor✅ Version

# Best Practices
print("=== Start Small ===")
print("""
1. CRITICAL COLUMNS FIRST
   - Not everything
   - Most important
   - Business-critical

2. FEW EXPECTATIONS
   - 5-10 per table
   - High value
   - Iterate

3. TEST EARLY
   - In dev
   - In CI
   - Before prod

4. ITERATE
   - Add more
   - Refine
   - Remove false positives
""")

print("\n=== Test in Pipelines ===")
print("""
1. INGESTION
   - Source validation
   - Schema check
   - Freshness

2. TRANSFORMATION
   - After each step
   - Before downstream

3. OUTPUT
   - Before consumers
   - Before ML

4. CONTINUOUS
   - Scheduled
   - On data arrival
   - Monitored
""")

print("\n=== Monitoring & Alerts ===")
print("""
1. DATA DOCS
   - Auto-generate
   - Host
   - Share

2. ALERTS
   - Slack
   - Email
   - PagerDuty

3. ON FAILURE
   - Fail pipeline
   - Quarantine
   - Alert

4. METRICS
   - Success rate
   - Trend
   - Dashboard
""")

print("\n=== Suite Management ===")
print("""
1. VERSION CONTROL
   - Git
   - Review
   - History

2. NAMING
   - Clear
   - Descriptive
   - Consistent

3. ORGANIZATION
   - By dataset
   - By domain
   - By layer

4. DOCUMENTATION
   - Why
   - What
   - Owner
""")

print("\n=== Anti-Patterns ===")
print("""
AVOID:
  ✗ Too many expectations
  ✗ No docs
  ✗ No alerts
  ✗ Tests in prod only
  ✗ Ignoring failures
  ✗ No versioning
  ✗ No ownership
  ✗ Duplicate expectations
""")

print("\n=== Checklist ===")
print("""
✓ Start with critical columns
✓ 5-10 expectations per table
✓ Test in dev + CI + prod
✓ Generate data docs
✓ Set up alerts
✓ Version control suites
✓ Document
✓ Monitor
✓ Iterate
✓ Share with stakeholders
""")

❓ FAQ

🎯 Frequently Asked Questions

Answers to common questions about Data Quality and Great Expectations.

What is Great Expectations?

Great Expectations (GX) is an open-source Python framework for data quality testing. It lets you define expectations about your data, validate against them, generate documentation, and integrate with pipelines to catch data issues before they reach production.

What are the 6 dimensions of data quality?

The six dimensions are: accuracy, completeness, consistency, timeliness, validity, and uniqueness. Great Expectations provides expectations for each dimension.

What is an expectation in Great Expectations?

An expectation is a declarative statement about your data, such as 'column X should never be null' or 'column Y should have values between 0 and 100'. Great Expectations has 300+ built-in expectations.

How does Great Expectations differ from dbt tests?

dbt tests are SQL-based and run inside your warehouse, focused on transformations. Great Expectations is Python-based, works with any data source (dataframes, SQL, Spark), and offers richer expectations, data docs, and profiling.

What is a data contract?

A data contract is a formal agreement between data producers and consumers that specifies schema, semantics, quality SLAs, and versioning. Great Expectations can enforce data contracts by validating data against defined expectations.

Is Great Expectations free?

Yes. Great Expectations is open source under Apache 2.0. There's also GX Cloud (commercial) for teams that want a managed platform.

What data sources does GX support?

GX supports: Pandas, Spark, SQL (Postgres, MySQL, Snowflake, BigQuery, Redshift), S3/GCS/Azure, Databricks, and custom sources.

Can I use GX with Airflow?

Yes. Use great_expectations_provider for a native Airflow operator. Run checkpoints as tasks, fail on validation failure, and get alerts.

What is a checkpoint in GX?

A checkpoint bundles data, expectation suites, and actions into a single unit. It's the primary way to run validations in production, with alerts, docs, and result storage.

How do I get started with GX?

Install with pip install great_expectations, run great_expectations init, connect a datasource, define expectations, and run a checkpoint. Start with 5-10 critical expectations and iterate.

# Quick Reference
print("=== Quick Answers ===")
print("""
Q: Is GX free?
A: Yes, Apache 2.0. GX Cloud is commercial.

Q: Do I need Spark?
A: No — works with Pandas, SQL, etc.

Q: How many expectations?
A: 300+ built-in. Custom too.

Q: Is it Python-only?
A: Yes. Python framework.

Q: Does it replace dbt tests?
A: Complements — use both.

Q: Learning curve?
A: Medium — 1-2 days to productive.

Q: Best first step?
A: 5 expectations on critical columns.
""")

🚀 Future

🎯 Where Data Quality Is Going

The future is AI-augmented, automated, real-time, and contract-driven. Data quality is becoming a first-class concern in every pipeline.

Future Directions

AI: Auto-generate. Contracts: Standard. Real-time: Streaming. Observability: Unified. Automated: Remediation. Open: Standards.

🚀 AI🚀 Contracts🚀 Real-Time🚀 Open

# Future of Data Quality
print("=== 1. AI-Augmented ===")
print("""
Trend: AI in data quality

AI:
  - Auto-generate expectations
  - Detect anomalies
  - Suggest fixes
  - Predict issues
  - Natural language

Future:
  - Self-tuning
  - Auto-remediation
  - Predictive
""")

print("\n=== 2. Data Contracts ===")
print("""
Trend: Contracts standard

Contracts:
  - Producer ↔ consumer
  - Schema
  - SLAs
  - Versioning
  - Enforced

Tools:
  - Data Contract CLI
  - GX
  - Custom

Future:
  - Industry standard
  - Automated
  - Compliant
""")

print("\n=== 3. Real-Time ===")
print("""
Trend: Streaming quality

Real-time:
  - Streaming data
  - Continuous checks
  - Low latency
  - On arrival

Tools:
  - GX + Kafka
  - Soda + Kafka
  - Custom

Future:
  - Real-time standard
  - Stream-native
  - Immediate alerts
""")

print("\n=== 4. Unified Observability ===")
print("""
Trend: Data + quality

Unified:
  - Lineage
  - Quality
  - Metrics
  - Logs
  - Alerts

Tools:
  - Monte Carlo
  - Datafold
  - Elementary
  - GX

Future:
  - One platform
  - Holistic view
  - Self-healing
""")

print("\n=== 5. Automated Remediation ===")
print("""
Trend: Auto-fix

Remediation:
  - Detect issue
  - Diagnose
  - Fix automatically
  - Verify

Examples:
  - Quarantine bad rows
  - Retry pipeline
  - Rollback
  - Alert on-call

Future:
  - Self-healing
  - Autonomous
  - Zero manual
""")

print("\n=== 6. Open Standards ===")
print("""
Trend: Open wins

Standards:
  - Open Data Contract
  - OpenLineage
  - OpenMetadata
  - GX suite format

Future:
  - Interoperability
  - No lock-in
  - Open ecosystem
""")

# TIMELINE
print("\n=== Timeline ===")
print("""
Past:
  2017: GX created
  2019: Growing adoption
  2021: GX Cloud
  2022: Data contracts
  2023: AI features

Present:
  2024: Mature framework
  2025: AI-augmented

Future:
  2030: Autonomous?
  2035: Standard?
  2040: New paradigms?
""")

print("\n=== Final Thoughts ===")
print("""
Data Quality is:
  - Critical
  - Measurable
  - Multi-dimensional
  - Continuous

Great Expectations is:
  - Python framework
  - 300+ expectations
  - Data Docs
  - Open source

The future is:
  - AI-augmented
  - Contract-driven
  - Real-time
  - Unified
  - Autonomous

"Trust your data.
 Test your expectations."

Keep exploring. Keep building.
""")