Une ligne écartée dans un déclaratif, c'est une sous-déclaration

La plupart des chaînes de qualité naissent de la même façon. Quelqu'un remarque des lignes portant une référence inconnue, ou un identifiant manquant. On ajoute une règle. La règle filtre. Et dans l'implémentation la plus simple, filtrer veut dire faire une jointure interne avec le référentiel.

Un déclaratif n'a pas le droit d'être incomplet

Dans la plupart des systèmes, écarter une ligne mal formée est une décision défendable. Dans un déclaratif réglementaire, non — et la différence mérite d'être dite franchement : l'obligation porte sur l'exhaustivité du périmètre. Une ligne qui n'arrive jamais dans le fichier n'est pas une ligne nettoyée. C'est une ligne non déclarée.

Ça change ce que la chaîne a le droit de faire. Elle peut différer une ligne, la signaler, la router vers un humain. Elle n'a pas le droit de la faire disparaître.

La jointure qui perd des lignes

Le référentiel se rafraîchit à son rythme. Les contrats arrivent au leur. Un titulaire enregistré le mardi après-midi existe dans le flux bien avant d'exister dans le référentiel. Écrivez la jointure comme ceci et le problème est invisible :

val enrichi = contrats.join(referentielPersonnes, Seq("id_personne"), "inner")

Tous les contrats rattachés à ce titulaire sortent de la chaîne. Ni signalés, ni journalisés, ni comptés. Et la supervision ne l'attrapera pas, parce qu'elle regarde la sortie — qui est parfaitement cohérente. Le fichier est bien formé, le schéma valide, toutes les règles sont au vert. Seul le total est un peu plus bas qu'il ne devrait, d'une fraction que personne ne questionne.

Un contrôle qui ne mesure que ce qu'il a gardé ne peut jamais voir ce qu'il a retiré.

Le mode de défaillance a un nom utile : la perte est silencieuse et corrélée. Silencieuse, parce qu'aucun compteur ne bouge. Corrélée, parce qu'elle ne frappe pas au hasard : elle frappe systématiquement les entités les plus récentes, celles qui viennent d'être créées. Autrement dit, exactement la population sur laquelle un contrôleur va poser des questions.

Écarter, jamais supprimer

Le correctif n'est pas une meilleure jointure. C'est un contrat différent : la chaîne a le droit de déplacer une ligne, jamais de la supprimer.

val enrichi = contrats.join(referentielPersonnes, Seq("id_personne"), "left")

val conformes = enrichi.filter(col("nom_titulaire").isNotNull)

val anomalies = enrichi
  .filter(col("nom_titulaire").isNull)
  .withColumn("regle",      lit("R27_TITULAIRE_NON_RATTACHE"))
  .withColumn("statut",     lit("A_TRAITER"))
  .withColumn("horodatage", current_timestamp())

La jointure externe garde tout. La séparation est explicite. Et la ligne porte le motif de sa mise à l'écart, ce qui la rend récupérable : dès que le référentiel rattrape son retard, un job planifié rejoue les lignes A_TRAITER et les solde.

Une précision qui a son importance en Spark : la jointure externe est plus coûteuse que l'interne, et l'optimiseur ne peut plus pousser certains filtres à travers elle. Sur des volumes importants, la forme qui tient est de calculer la séparation une seule fois, en marquant les lignes, puis de dériver les deux sorties depuis le même DataFrame mis en cache — sinon la jointure est évaluée deux fois.

val marque = enrichi.withColumn("conforme", col("nom_titulaire").isNotNull).cache()

val conformes = marque.filter(col("conforme"))
val anomalies = marque.filter(!col("conforme"))

Le contrôle qui rend la perte impossible

Une seule assertion, à la fin de chaque run, rend toute cette famille de défauts impossible :

val collecte  = brut.count()
val declare   = conformes.count()
val ecartees  = anomalies.count()

require(
  collecte == declare + ecartees,
  s"perte silencieuse : ${collecte - declare - ecartees} lignes"
)

Elle coûte trois comptages par run. Sa valeur, c'est qu'elle fait échouer le job plutôt que le déclaratif : une chaîne qui s'arrête à 01 h 12, c'est un incident ; une chaîne qui dépose un fichier incomplet, c'est un constat. Ces deux-là n'ont pas le même prix.

Elle attrape aussi ce que les tests unitaires ne voient pas. Chaque fonction était juste isolément ; le défaut vivait dans le raccord entre deux d'entre elles. Les contrôles d'équilibre sont les seuls tests qui regardent les raccords — et c'est pour ça qu'ils doivent tourner en production, pas seulement en intégration.

Trois pièges classiques sur cette assertion. Elle doit compter les lignes reçues, pas les lignes lues après un éventuel dropDuplicates. Elle doit tourner après la dernière écriture, pas avant. Et elle doit lire les compteurs depuis les tables écrites, pas depuis les DataFrames en mémoire : sinon elle valide votre intention, pas votre résultat.

L'anomalie est un objet métier

La dernière étape est celle qu'on saute d'habitude. Une quarantaine dont rien ne sort est une poubelle mieux nommée. Les lignes n'en sortent que si quelqu'un peut les voir, comprendre pourquoi elles y sont, et agir — donc un écran, pas un fichier de log.

• Un code de règle par ligne. Agrégez les motifs et vous perdez la capacité de corriger la cause.
• Une alerte sur l'ancienneté du stock. Une ligne en attente depuis trois semaines, c'est un processus arrêté, pas une ligne qui patiente.
• Le compteur d'anomalies sur le tableau de bord métier. Pas dans les logs techniques. C'est le chiffre qui dit si le référentiel suit.
• Un taux de sortie, pas seulement un stock. Combien de lignes sont entrées en quarantaine cette semaine, combien en sont sorties. Si le second nombre est nul pendant un mois, la boucle de correction n'existe que sur le papier.

Si vous ne retenez qu'une chose : allez lire les jointures de votre couche qualité, et comptez combien sont internes. Puis demandez-vous ce que deviennent les lignes qu'elles écartent.