Data Quality Engineer

Download SKILL.mdGet Murage How skills work

Open source. macOS, Windows and Linux.

At a glance

What it is

Data quality engineering covering Great Expectations suite design, data contracts between teams, schema validation at ingestion, statistical anomaly detection, automated data profiling, quality scoring frameworks, SLA monitoring, freshness checks, and remediation workflows for maintaining trustworthy data across the organization.

A skill is a written procedure a bot loads when a job calls for it. This one is a single file, SKILL.md, and the whole file is on this page.

File: SKILL.md
Length: 546 lines1,792 words, about 9 min read
Category: Data engineeringpipelines and ETL
Level: Intermediate
License

More data engineering skills.

The full skill.

The whole file, exactly as it downloads. Read it before you rely on it.

SKILL.md

546 lines · 1,792 words

Download

Data Quality Engineer

Overview

Data quality engineering is the practice of building systems and processes that ensure data is accurate, complete, consistent, timely, and fit for purpose. This goes beyond one-off validation to encompass continuous monitoring, contractual agreements between data producers and consumers, automated anomaly detection, and organizational data quality culture.

Data Quality Dimensions

Dimension        Measure                           Example Check
---------------------------------------------------------------------------
Accuracy         Values match real-world truth      Address geocodes within bounds
Completeness     Required fields are populated      customer_email NOT NULL rate > 99%
Consistency      Same facts agree across systems    CRM revenue = billing revenue
Timeliness       Data arrives within SLA            Daily feed lands by 06:00 UTC
Uniqueness       No unwanted duplicates             order_id is unique per day
Validity         Values conform to rules            status IN ('active','inactive')
Freshness        Data is recent enough              max(updated_at) < 2 hours ago
Integrity        Referential relationships hold     Every order.customer_id exists

Put this skill to work.

Questions.

What is the Data Quality Engineer skill?

How much does it cost?

Nothing. It ships with Murage, which is open source. Your bots run on the AI plan you already have.

Can I change it?

Yes. Open it in Settings → Skills to read, edit or duplicate it. Skill Guard scans it again after every edit.

Give your first job to Murage.

Download the app, connect the AI you already pay for, and tell your Chief of Staff what needs doing. Plan on about ten minutes from install to a working team.

Download Murage

No account needed · Runs on the AI plan you already pay for

Dimension Measure Example Check --------------------------------------------------------------------------- Accuracy Values match real-world truth Address geocodes within bounds Completeness Required fields are populated customer_email NOT NULL rate > 99% Consistency Same facts agree across systems CRM revenue = billing revenue Timeliness Data arrives within SLA Daily feed lands by 06:00 UTC Uniqueness No unwanted duplicates order_id is unique per day Validity Values conform to rules status IN ('active','inactive') Freshness Data is recent enough max(updated_at) < 2 hours ago Integrity Referential relationships hold Every order.customer_id exists

import numpy as np from scipy import stats class DataAnomalyDetector: def __init__(self, history_days=90, sensitivity=3.0): self.history_days = history_days self.sensitivity = sensitivity def detect_volume_anomaly(self, current_count, historical_counts): """Z-score based volume anomaly detection.""" mean = np.mean(historical_counts) std = np.std(historical_counts) if std == 0: return False, 0.0 z_score = (current_count - mean) / std is_anomaly = abs(z_score) > self.sensitivity return is_anomaly, z_score def detect_distribution_shift(self, current_values, historical_values): """Kolmogorov-Smirnov test for distribution changes.""" statistic, p_value = stats.ks_2samp(current_values, historical_values) is_anomaly = p_value < 0.01 # 1% significance level return is_anomaly, {"statistic": statistic, "p_value": p_value} def detect_null_rate_anomaly(self, column_name, current_null_pct, historical_null_pcts): """Alert when null rate deviates from historical pattern.""" mean_null = np.mean(historical_null_pcts) std_null = np.std(historical_null_pcts) if std_null == 0: is_anomaly = current_null_pct != mean_null else: z = (current_null_pct - mean_null) / std_null is_anomaly = abs(z) > self.sensitivity return is_anomaly, { "column": column_name, "current": current_null_pct, "expected": mean_null, } def detect_cardinality_anomaly(self, current_unique, historical_uniques): """Detect sudden changes in column cardinality.""" median_unique = np.median(historical_uniques) mad = np.median(np.abs(historical_uniques - median_unique)) if mad == 0: mad = 1.0 modified_z = 0.6745 * (current_unique - median_unique) / mad is_anomaly = abs(modified_z) > self.sensitivity return is_anomaly, modified_z

from dataclasses import dataclass from typing import Dict, List @dataclass class ColumnProfile: name: str dtype: str count: int null_count: int null_pct: float unique_count: int unique_pct: float min_value: any max_value: any mean: float = None median: float = None std: float = None p5: float = None p95: float = None top_values: List[tuple] = None pattern_matches: Dict[str, float] = None class DataProfiler: def profile_dataframe(self, df) -> Dict[str, ColumnProfile]: profiles = {} for col in df.columns: profiles[col] = self._profile_column(df, col) return profiles def _profile_column(self, df, col_name): col = df[col_name] profile = ColumnProfile( name=col_name, dtype=str(col.dtype), count=len(col), null_count=col.isnull().sum(), null_pct=col.isnull().mean() * 100, unique_count=col.nunique(), unique_pct=col.nunique() / len(col) * 100 if len(col) > 0 else 0, min_value=col.min() if not col.isnull().all() else None, max_value=col.max() if not col.isnull().all() else None, ) if col.dtype in ['int64', 'float64']: profile.mean = col.mean() profile.median = col.median() profile.std = col.std() profile.p5 = col.quantile(0.05) profile.p95 = col.quantile(0.95) if col.dtype == 'object': profile.top_values = col.value_counts().head(10).to_dict() profile.pattern_matches = self._detect_patterns(col) return profile def _detect_patterns(self, col): patterns = { 'email': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', 'phone': r'^\+?[\d\s\-]{7,15}$', 'uuid': r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', 'date_iso': r'^\d{4}-\d{2}-\d{2}$', 'url': r'^https?://[^\s]+$', } results = {} non_null = col.dropna() for name, pattern in patterns.items(): match_pct = non_null.str.match(pattern).mean() * 100 if match_pct > 50: results[name] = match_pct return results def compare_profiles(self, profile_a, profile_b) -> List[str]: """Compare two profiles and report significant differences.""" alerts = [] for col_name in profile_a: if col_name not in profile_b: alerts.append(f"Column '{col_name}' missing in new data") continue a, b = profile_a[col_name], profile_b[col_name] if abs(a.null_pct - b.null_pct) > 5: alerts.append(f"'{col_name}' null rate changed: {a.null_pct:.1f}% -> {b.null_pct:.1f}%") if a.unique_pct > 90 and b.unique_pct < 50: alerts.append(f"'{col_name}' cardinality dropped: {a.unique_pct:.0f}% -> {b.unique_pct:.0f}%") return alerts

Data Quality Engineer

More data engineering skills.

The full skill.

Data Quality Engineer

Overview

Data Quality Dimensions

Put this skill to work.

Questions.

Give your first job to Murage.

Search murage.ai

Data Quality Engineer

More data engineering skills.

The full skill.

Data Quality Engineer

Overview

Data Quality Dimensions

Put this skill to work.

Questions.

Give your first job to Murage.

Create a bot and use this skill.

Ask Ember to create the bot

Open the bot’s skills

Find Data Quality Engineer and switch it on

Ask the bot for the job

About this skill.

Great Expectations: Production Patterns

Multi-Layer Validation Architecture

Custom Expectations

Data Contracts

Contract Specification

Contract Enforcement in Pipeline

Anomaly Detection

Statistical Anomaly Detection

Data Profiling

Automated Profiling Pipeline

Quality Scoring Framework

Data Quality Remediation Workflow

Checklist: Data Quality Program Setup

When to Use

Output Format

Example

Edge Cases