Data quality & testing: Great Expectations, Soda, dbt tests, data validation, freshness monitoring, anomaly detection, data observability, schema drift.
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
Data quality & testing: Great Expectations, Soda, dbt tests, data validation, freshness monitoring, anomaly detection, data observability, schema drift.
import great_expectations as ge
# Créer une suite d'expectations
suite = context.add_expectation_suite("sensors_quality_suite")
batch_request = BatchRequest(
datasource_name="sensors_datasource",
data_connector_name="parquet_connector",
data_asset_name="sensor_readings",
)
validator = context.get_validator(
batch_request=batch_request,
expectation_suite_name="sensors_quality_suite",
)
# Expectations sur les colonnes
validator.expect_column_to_exist("machine_id")
validator.expect_column_to_exist("temperature")
validator.expect_column_to_exist("pression")
validator.expect_column_to_exist("timestamp")
# Contraintes de nullité
validator.expect_column_values_to_not_be_null("machine_id")
validator.expect_column_values_to_not_be_null("temperature")
validator.expect_column_values_to_not_be_null("timestamp")
# Contraintes de plage
validator.expect_column_values_to_be_between(
"temperature", min_value=-50, max_value=300
)
validator.expect_column_values_to_be_between(
"pression", min_value=0, max_value=50
)
# Contraintes de type
validator.expect_column_values_to_be_of_type("machine_id", "str")
validator.expect_column_values_to_be_of_type("temperature", "float64")
# Contraintes d'unicité (clé composite)
validator.expect_compound_columns_to_be_unique(
["machine_id", "timestamp"]
)
# Distribution (quantiles)
validator.expect_column_quantile_values_to_be_between(
"temperature",
quantile_ranges={
"quantiles": [0.05, 0.25, 0.5, 0.75, 0.95],
"value_ranges": [
[10, 30], # 5% des valeurs entre 10 et 30
[40, 60], # 25% des valeurs entre 40 et 60
[55, 75], # 50% (médiane) entre 55 et 75
[70, 90], # 75% des valeurs entre 70 et 90
[85, 120], # 95% des valeurs entre 85 et 120
],
}
)
# Valeurs distinctes
validator.expect_column_distinct_values_to_be_in_set(
"alerte", ["NORMAL", "WARNING", "CRITIQUE"]
)
# Taille de la table
validator.expect_table_row_count_to_be_between(min_value=100, max_value=50000)
# Enregistrer la suite
validator.save_expectation_suite(discard_failed_expectations=False)
# Générer les Data Docs
context.build_data_docs()
1.4 Exécution des Validations
# Valider un nouveau batch
results = context.run_checkpoint(
checkpoint_name="sensors_checkpoint",
batch_request=batch_request,
)
# Résultatsif results["success"]:
print("✅ Toutes les expectations passent")
else:
for result in results["run_results"].values():
for exp_result in result["expectation_suite_results"]:
ifnot exp_result["success"]:
print(f"❌ Échec : {exp_result['expectation_config']['expectation_type']} "f"— {exp_result.get('exception_info', {}).get('raised_exception', '')}")
1.5 Data Docs (HTML)
# Générer et ouvrir les rapports
great_expectations docs build
great_expectations docs open
defdetect_late_data(table_path: str, max_lag_hours: int = 2) -> dict:
"""Vérifie que les données les plus récentes ne sont pas trop vieilles"""
df = pd.read_parquet(table_path)
max_ts = df['timestamp'].max()
now = datetime.utcnow()
lag = (now - max_ts).total_seconds() / 3600
result = {
"table": table_path,
"latest_timestamp": str(max_ts),
"lag_hours": round(lag, 2),
"status": "ok"if lag <= max_lag_hours else"critical",
}
if lag > max_lag_hours:
result["alert"] = f"Données en retard : {lag:.1f}h (max {max_lag_hours}h)"elif lag > max_lag_hours * 0.8:
result["status"] = "warning"return result
# SQL pour tableau de bord qualitéSELECTtable_name,last_checked_at,ROUND(freshness_minutes,1)ASfreshness_min,row_count,null_rate,duplicate_rate,schema_version,CASEWHENfreshness_minutes>60THEN'🔴'WHENfreshness_minutes>30THEN'🟡'ELSE'🟢'ENDASfreshness_status,CASEWHENnull_rate>0.05THEN'🔴'WHENnull_rate>0.01THEN'🟡'ELSE'🟢'ENDASnull_statusFROMdata_quality.metricsWHEREdate=CURRENT_DATEORDERBYtable_name;
6.2 Intégration Airflow
from airflow import DAG
from airflow.operators.python import PythonOperator
defcheck_data_quality(**context):
import great_expectations as ge
from great_expectations.data_context import FileDataContext
context_gx = FileDataContext(project_root_dir="./gx_project")
results = context_gx.run_checkpoint(
checkpoint_name="sensors_checkpoint",
batch_request=None,
)
ifnot results["success"]:
raise ValueError("Échec des contrôles de qualité des données")
with DAG(...) as dag:
extract = PythonOperator(...)
transform = PythonOperator(...)
load = PythonOperator(...)
quality_check = PythonOperator(
task_id='data_quality_check',
python_callable=check_data_quality,
)
load >> quality_check # Le quality gate bloque la suite
Pièges Courants (Pitfalls)
Trop de tests, pas assez d'actions.
Erreur : 200 expectations qui échouent toutes → les équipes ignorent les alertes.
Correction : Commencer par 5-10 tests critiques (nulls sur clés, freshness, doublons). Ajouter progressivement.
Seuils trop stricts.
Erreur :expect_column_values_to_be_between("temperature", 10, 50) → bloque des données valides la nuit où il fait 8°C.
Correction : Utiliser des seuils adaptés à la réalité terrain. Ajouter des fenêtres de tolérance.
Pas de gestion des valeurs manquantes autorisées.
Erreur :expect_column_values_to_not_be_null sur une colonne optionnelle → blocage du pipeline.
Correction :mostly=0.95 pour tolérer 5% de nulls.
Freshness mal configurée sur des pipelines à latence.
Erreur : Un pipeline qui met 45 minutes à s'exécuter → alerte freshness fausse toutes les heures.
Correction : Configurer le seuil en fonction du SLA réel du pipeline, pas d'un idéal.
Schema drift ignoré.
Erreur : Une source ajoute une colonne → les validations échouent silencieusement.
Correction : Configurer when schema changes: WARN dans Soda, et investiguer automatiquement.
Liste de Vérification (Checklist)
Great Expectations ou Soda configuré sur chaque couche (bronze/silver/gold).
Tests de nullité sur les colonnes clés (primary key, FK).
Tests de plage sur les mesures numériques.
Freshness monitoring configuré (< 30 min pour temps réel, < 24h pour batch).
Détection de doublons sur les clés composites.
Schema drift détecté et alerté.
Anomaly detection sur les volumes (Z-score > 3 ou écart > 20%).
Pipeline de qualité bloquant (quality gate avant utilisation des données).
