Data quality & testing: Great Expectations, Soda, dbt tests, data validation, freshness monitoring, anomaly detection, data observability, schema drift.
Instrucciones de origen · Vista previa de solo lectura
name
data-quality-testing
description
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).
