Data Quality Monitoring and Trending

Organizations need to track data quality metrics over time, identify trends, and proactively detect degradation. Understanding how to monitor and visualize data quality trends is essential.

You Might Be Asking:

• How do I track data quality over time?
• Can I see historical assertion pass rates?
• How do I create quality dashboards?

1. Create assertions with historical tracking:

# Assertions automatically track history
# Access via UI: Dataset → Quality → Assertion → History

# Via GraphQL:
query getAssertionHistory {
  assertion(urn: "urn:li:assertion:12345") {
    runEvents(start: 0, count: 100) {
      total
      runEvents {
        timestampMillis
        status
        result {
          type
          actualAggValue
          externalUrl
        }
      }
    }
  }
}

2. Calculate quality scores:

from datahub.emitter.rest_emitter import DatahubRestEmitter
import requests

def calculate_dataset_quality_score(dataset_urn, lookback_days=30):
    """
    Calculate quality score based on assertion pass rate
    """
    query = """
    query getDatasetAssertions($urn: String!) {
      dataset(urn: $urn) {
        assertions(start: 0, count: 100) {
          assertions {
            urn
            runEvents(limit: 100) {
              runEvents {
                status
                timestampMillis
              }
            }
          }
        }
      }
    }
    """

    # Fetch assertion results
    response = requests.post(
        "http://localhost:8080/api/graphql",
        json={"query": query, "variables": {"urn": dataset_urn}},
        headers={"Authorization": f"Bearer {TOKEN}"}
    )

    assertions = response.json()['data']['dataset']['assertions']['assertions']

    # Calculate pass rate
    total_runs = 0
    passed_runs = 0

    cutoff_time = time.time() * 1000 - (lookback_days * 24 * 60 * 60 * 1000)

    for assertion in assertions:
        for run in assertion['runEvents']['runEvents']:
            if run['timestampMillis'] > cutoff_time:
                total_runs += 1
                if run['status'] == 'COMPLETE':
                    passed_runs += 1

    quality_score = (passed_runs / total_runs * 100) if total_runs > 0 else 0
    return quality_score

# Usage
score = calculate_dataset_quality_score(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table,PROD)",
    lookback_days=30
)
print(f"Quality Score: {score:.2f}%")

3. Set up quality monitoring dashboard:

# Export quality metrics for external dashboarding
import pandas as pd

def export_quality_metrics(datasets, output_file):
    """
    Export quality metrics for multiple datasets
    """
    metrics = []

    for dataset_urn in datasets:
        # Get assertion results
        score = calculate_dataset_quality_score(dataset_urn)

        # Get dataset details
        dataset_info = get_dataset_info(dataset_urn)

        metrics.append({
            'dataset_name': dataset_info['name'],
            'platform': dataset_info['platform'],
            'quality_score': score,
            'assertion_count': dataset_info['assertion_count'],
            'last_updated': dataset_info['last_modified'],
            'owners': ', '.join(dataset_info['owners'])
        })

    df = pd.DataFrame(metrics)
    df.to_csv(output_file, index=False)
    return df

# Create quality report
df = export_quality_metrics(all_datasets, 'quality_report.csv')

# Can be imported to Looker, Tableau, or other BI tools

4. Alert on quality degradation:

Example for Slack alerting. This can be adapted for other notification platforms like Microsoft Teams, email, or PagerDuty.

# DataHub Actions config for quality alerting
action:
  type: "quality_alert"
  config:
    # Monitor assertion pass rate
    threshold:
      min_pass_rate: 95.0  # Alert if < 95%
      window_hours: 24

    # Alert on consecutive failures
    consecutive_failures: 3

    # Notification
    slack_webhook: "${SLACK_WEBHOOK}"
    message_template: |
      ⚠️ Data Quality Alert
      Dataset: {{dataset.name}}
      Quality Score: {{quality_score}}%
      Failed Assertions: {{failed_count}}/{{total_count}}

      Recent Failures:
      {{#each failures}}
      - {{assertion.name}}: {{result.message}}
      {{/each}}

5. Track quality trends over time:

# Generate quality trend report
def generate_quality_trend(dataset_urn, days=90):
    """
    Generate daily quality scores for trending
    """
    daily_scores = {}

    # Get all assertion runs for period
    query = """
    query getAssertionRuns($urn: String!) {
      dataset(urn: $urn) {
        assertions {
          assertions {
            runEvents(limit: 1000) {
              runEvents {
                status
                timestampMillis
              }
            }
          }
        }
      }
    }
    """

    # Fetch and aggregate by day
    # Calculate daily pass rates
    # Return time series data

    return daily_scores

# Visualize trend
import matplotlib.pyplot as plt

scores = generate_quality_trend(dataset_urn, days=90)
dates = list(scores.keys())
values = list(scores.values())

plt.figure(figsize=(12, 6))
plt.plot(dates, values)
plt.axhline(y=95, color='r', linestyle='--', label='SLA Threshold')
plt.title('Data Quality Trend (90 days)')
plt.xlabel('Date')
plt.ylabel('Quality Score (%)')
plt.legend()
plt.savefig('quality_trend.png')

6. Create quality health check report:

# Daily quality health check
def daily_quality_report():
    """
    Generate daily quality health check across all critical datasets
    """
    report = {
        'date': datetime.now().strftime('%Y-%m-%d'),
        'datasets': []
    }

    critical_datasets = get_critical_datasets()

    for dataset in critical_datasets:
        quality_score = calculate_dataset_quality_score(dataset)

        status = 'HEALTHY'
        if quality_score < 95:
            status = 'WARNING'
        if quality_score < 90:
            status = 'CRITICAL'

        report['datasets'].append({
            'name': dataset,
            'score': quality_score,
            'status': status
        })

    # Send summary
    summary = f"""
    Data Quality Daily Report - {report['date']}

    Total Datasets: {len(report['datasets'])}
    Healthy: {sum(1 for d in report['datasets'] if d['status'] == 'HEALTHY')}
    Warning: {sum(1 for d in report['datasets'] if d['status'] == 'WARNING')}
    Critical: {sum(1 for d in report['datasets'] if d['status'] == 'CRITICAL')}

    Average Quality Score: {sum(d['score'] for d in report['datasets']) / len(report['datasets']):.2f}%
    """

    send_slack_message(summary)
    return report

Quality monitoring requires assertions to be set up and running regularly. Export metrics to your preferred BI tool for richer visualizations. Establish quality SLAs and alert thresholds. Review quality trends in regular data governance meetings.

• Assertions
• Data Quality Guide