메뉴
Data pipelines: conception, patterns ETL/ELT, médaillon architecture (bronze/silver/gold), data contracts, lineage, métriques, tests, reverse ETL.
기본적으로 소스를 먼저 확인하는 Prompt가 선택됩니다. 직접 명령으로 전환하거나 로컬 사본을 다운로드할 수도 있습니다.
설치 여부를 결정하기 전에 SKILL.md와 SkillsMP에 표시된 보조 파일을 읽어 보세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/JohnNuwan/EVA_CORE --skill data-pipelines명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
Concevoir et maintenir un watchdog auto-correcteur pour services HTTP — checks de santé, auto-restart, état persistant, rapports et pièges bash.
Serveur de messagerie sécurisé auto-hébergé (Signal-like) avec Flask + WebSocket + AES-256-GCM + pont EVA
ADAM-SENTINEL — Veilleur technologique 24h/24h. Scanne 10 domaines, cree des rapports, met a jour les skills, alerte sur les CVE et breaking changes.
SOC 직업 분류 기준
SKILL.md 표시 중
| name | data-pipelines |
| description | Data pipelines: conception, patterns ETL/ELT, médaillon architecture (bronze/silver/gold), data contracts, lineage, métriques, tests, reverse ETL. |
| version | 1.0.0 |
| author | EVA & The Hive |
| license | Privée EVA St-Étienne |
| platforms | ["linux","macos","windows"] |
| metadata | {"EVA":{"tags":["data-pipeline","etl","elt","medallion-architecture","data-contract","data-lineage","data-observability","reverse-etl","dataops"],"homepage":"https://dataprocess.website/","related_skills":["airflow-orchestration","apache-spark","apache-kafka","data-warehouse","data-lakes-lakehouse","dbt-analytics","data-quality-testing"]}} |
| prerequisites | {"commands":["python3","pip"],"pip_packages":["pandas","pyarrow","great_expectations","soda-core","dbt-core","sqlalchemy","psycopg2-binary","openlineage-python"]} |
Un data pipeline est l'épine dorsale de toute organisation data-driven. Cette compétence couvre l'architecture complète :
Activez cette compétence lorsque l'utilisateur :
┌─────────────────────────────────────────────────────────────┐
│ GOLD (Agrégé) │
│ KPIs, OEE, rapports, tableaux de bord, ML features │
│ Modèle dimensionnel, star schema, vues matérialisées │
├─────────────────────────────────────────────────────────────┤
│ SILVER (Nettoyé) │
│ Données dédupliquées, typées, validées, enrichies │
│ Jointures, normalisation, SCD type 2 │
├─────────────────────────────────────────────────────────────┤
│ BRONZE (Raw) │
│ Copie exacte des sources, format natif (JSON/Avro/Parquet) │
│ Partitionné par date, immuable, rétention configurable │
└─────────────────────────────────────────────────────────────┘
# Ingestion depuis Kafka vers bronze (Delta/Parquet)
import pandas as pd
from datetime import datetime
def ():
df = pd.read_parquet(source_path)
df[] = datetime.utcnow()
df[] = source_path
date_str = datetime.utcnow().strftime()
output =
df.to_parquet(output, partition_cols=[], compression=)
output
def bronze_to_silver(bronze_path: str, silver_path: str):
"""Déduplication, typage, validation"""
df = pd.read_parquet(bronze_path)
# Nettoyage
df = df.drop_duplicates(subset=["machine_id", "timestamp"])
df = df.dropna(subset=["machine_id", "temperature"])
# Typage strict
df['temperature'] = pd.to_numeric(df['temperature'], errors='coerce')
df['timestamp'] = pd.to_datetime(df['timestamp'], errors='coerce')
df = df[df['temperature'].between(-50, 300)] # Plage réaliste
# Enrichissement
df['date'] = df['timestamp'].dt.date
df['heure'] = df['timestamp'].dt.hour
df['jour_semaine'] = df['timestamp'].dt.dayofweek
# SCD Type 1 (écrasement)
df.to_parquet(
f"{silver_path}/sensors/date={df['date'].iloc[0]}/",
partition_cols=["date"],
compression="snappy"
)
return len(df)
# Caractéristiques silver :
# - Données propres, typées, validées
# - Pas de doublons
# - Colonnes standardisées
# - SCD type 1 ou 2 selon l'usage
def silver_to_gold(silver_path: str, gold_path: str):
"""Agrégations métier et KPIs"""
df = pd.read_parquet(silver_path)
# Agrégation machine + heure
df_gold = df.groupby(["machine_id", "date", "heure"]).agg(
temp_moyenne=("temperature", "mean"),
temp_max=("temperature", "max"),
temp_min=("temperature", "min"),
temp_std=("temperature", "std"),
nb_mesures=("temperature", "count"),
nb_alertes=("alerte", lambda x: (x == "CRITIQUE").sum()),
).reset_index()
df_gold.to_parquet(
f"{gold_path}/kpi_sensors/date={df['date'].iloc[0]}/",
partition_cols=["date"],
compression="snappy"
)
return df_gold.shape[0]
Avantages :
- Données nettoyées avant chargement → base cible légère
- Moins de compute sur la cible
- Idéal pour sources sensibles / PII à anonymiser
Inconvénients :
- Transform engine doit scaler (Spark)
- Perte de granularité si la cible ne stocke pas le raw
- Plus lent à l'ingestion
Avantages :
- Ingestion rapide (copie raw)
- Transform engine = la cible (Snowflake, BigQuery, Redshift)
- Toute la flexibilité des transformations post-hoc
- Data scientists accèdent aux données brutes
Inconvénients :
- Compute coûteux sur la cible
- Nécessite un warehouse scalable
- Données raw = stockage volumineux
| Critère | ETL | ELT |
|---|---|---|
| Volume de données | Moyen (< 10 To) | Très grand (> 10 To) |
| Cible | Base traditionnelle | Data Warehouse Cloud |
| Latence requise | Modérée | Faible à modérée |
| Data Science raw | Non | Oui |
| Anonymisation | Avant chargement | Possible après |
| Coût compute | Fixe (cluster Spark) | Variable (warehouse) |
Un data contract est un accord formel entre un producteur et un consommateur de données, spécifiant :
# contracts/sensor_readings.yaml
dataset: sensors.bronze.sensor_readings
owners:
producer: team-ingestion
consumer: team-analytics
version: 1.2.0
schema:
fields:
machine_id:
type: string
required: true
description: Identifiant unique de la machine
temperature:
type: double
required: true
constraints:
min: -50
max: 300
pression:
type: double
required: false
constraints:
min: 0
max: 50
timestamp:
type: timestamp
required: true
primary_key: [machine_id, timestamp]
freshness:
interval: 15 minutes
sla: 5 minutes
volume:
min_rows_per_hour: 1000
max_rows_per_hour: 100000
quality:
null_rate_max: 0.01
duplicate_rate_max: 0.0
validity_rate_min: 0.98
import yaml
import pandas as pd
def validate_data_contract(df: pd.DataFrame, contract_path: str) -> dict:
with open(contract_path) as f:
contract = yaml.safe_load(f)
results = {"passed": True, "checks": []}
# Validation schema
for field, spec in contract['schema']['fields'].items():
if spec['required'] and field not in df.columns:
results['checks'].append(f"MISSING: {field}")
results['passed'] = False
# Validation contraintes
for field, spec in contract['schema']['fields'].items():
if 'constraints' in spec and field in df.columns:
if 'min' in spec['constraints']:
if df[field].min() < spec['constraints']['min']:
results['checks'].append(f"MIN_FAIL: {field}")
results['passed'] = False
return results
from openlineage.client import OpenLineageClient
from openlineage.client.run import RunEvent, RunState, Run
from openlineage.client.event import Dataset, Job, ParentRun
from datetime import datetime
client = OpenLineageClient(url="http://marquez:5000")
def emit_lineage(job_name: str, inputs: list, outputs: list):
"""Émet un événement de lineage"""
event = RunEvent(
eventType=RunState.COMPLETE,
eventTime=datetime.utcnow().isoformat(),
run=Run(runId=job_name),
job=Job(namespace="eva-pipelines", name=job_name),
inputs=[Dataset(namespace="s3", name=p) for p in inputs],
outputs=[Dataset(namespace="s3", name=p) for p in outputs],
producer="eva-data-pipeline"
)
client.emit(event)
# Usage
emit_lineage(
"bronze_to_silver",
inputs=["s3://data-lake/bronze/sensors/date=2026-07-22/"],
outputs=["s3://data-lake/silver/sensors/date=2026-07-22/"]
)
docker run -d -p 5000:5000 -p 5001:5001 marquezproject/marquez:latest
# UI : http://localhost:3000
| Métrique | Description | Seuil d'alerte |
|---|---|---|
| Freshness | Temps depuis la dernière mise à jour | > 30 min |
| Volume | Nombre de lignes par intervalle | ±20% vs moyenne |
| Schema Drift | Changement dans les colonnes / types | Tout changement |
| Null Rate | Ratio de valeurs manquantes | > 1% sur colonnes clés |
| Row Count | Nombre de lignes écrites | < seuil minimum |
import time
class PipelineMonitor:
def __init__(self, pipeline_name: str):
self.pipeline_name = pipeline_name
self.start_time = time.time()
self.metrics = {}
def record_row_count(self, table: str, count: int):
self.metrics[f"{table}.row_count"] = count
# Alerte si volume anormal
expected = self._get_expected_volume(table)
if expected and abs(count - expected) / expected > 0.2:
print(f"⚠️ Volume anormal sur {table}: {count} (attendu: {expected})")
def record_freshness(self, table: str, last_timestamp):
lag = datetime.now() - last_timestamp
self.metrics[f"{table}.freshness_min"] = lag.total_seconds() / 60
if lag > timedelta(minutes=30):
print(f"🔴 Freshness dépassée sur {table}: {lag}")
def finish(self):
duration = time.time() - self.start_time
self.metrics["duration_s"] = duration
# Envoyer vers InfluxDB / Prometheus
return self.metrics
# tests/test_sensors_pipeline.py
import pytest
import pandas as pd
def test_bronze_schema():
"""Vérifie que les colonnes bronze sont présentes"""
df = pd.read_parquet("/tmp/test_bronze/")
required_cols = {"machine_id", "temperature", "timestamp", "_ingested_at"}
assert required_cols.issubset(df.columns), f"Manque: {required_cols - set(df.columns)}"
def test_silver_clean():
"""Vérifie l'absence de doublons dans silver"""
df = pd.read_parquet("/tmp/test_silver/")
dupes = df.duplicated(subset=["machine_id", "timestamp"]).sum()
assert dupes == 0, f"{dupes} doublons trouvés"
def test_gold_kpis():
"""Vérifie les KPIs gold"""
df = pd.read_parquet("/tmp/test_gold/")
assert df['temp_moyenne'].between(-50, 300).all()
assert df['nb_mesures'].min() >= 0
def validate_pipeline_output(stage: str, path: str) -> bool:
"""Checklist de validation après chaque étape"""
import pyarrow.parquet as pq
pf = pq.ParquetFile(path)
schema = pf.schema
checks = {
"bronze": lambda s: "_ingested_at" in str(s),
"silver": lambda s: "temperature" in str(s) and "heure" in str(s),
"gold": lambda s: "temp_moyenne" in str(s),
}
if stage in checks:
result = checks[stage](schema)
if not result:
print(f"ÉCHEC validation {stage}")
return False
return True
def reverse_etl_to_crm(metrics_path: str, api_endpoint: str):
"""Envoie les KPIs gold vers le CRM/Slack/ERP"""
df = pd.read_parquet(metrics_path)
# Agrégation quotidienne par machine
daily_kpis = df.groupby("machine_id").agg(
oee=("oee", "mean"),
alert_count=("nb_alertes", "sum"),
).reset_index()
# Push vers API CRM
import requests
for _, row in daily_kpis.iterrows():
requests.post(api_endpoint, json={
"machine_id": row['machine_id'],
"oee_score": round(row['oee'], 3),
"alert_count": int(row['alert_count']),
})
Pipeline non idempotent.
Pas de gestion des données tardives (late data).
Manque de monitoring de la fraîcheur.
max(timestamp) de chaque table.Schema drift non détecté.
Pipeline trop rigide (hardcoded paths).