A row set aside in a regulatory filing is an under-declaration

Most data quality pipelines start the same way. Someone notices rows carrying an unknown reference, or a missing identifier. A rule is added. The rule filters. And in the simplest implementation, filtering means an inner join against the reference table.

A filing has no right to be incomplete

In most systems, dropping a malformed row is a defensible decision. In a regulatory filing it is not — and the difference is worth stating plainly: the obligation is completeness of scope. A row that never reaches the file is not a cleaned row. It is a row that was not declared.

That changes what the pipeline is allowed to do. It may defer a row, flag it, route it to a human. It may not make it disappear.

The join that loses rows

Reference data refreshes on its own schedule. Contracts arrive on theirs. A holder registered on Tuesday afternoon exists in the feed well before it exists in the reference table. Write the join like this and the problem is invisible:

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

Every contract attached to that holder leaves the chain. Not flagged, not logged, not counted. And the monitoring will not catch it, because monitoring watches the output — which is perfectly consistent. The file is well formed, the schema validates, every rule is green. Only the total is slightly lower than it should be, by a fraction nobody questions.

A control that measures only what it kept can never see what it removed.

The failure mode has a useful name: the loss is silent and correlated. Silent, because no counter moves. Correlated, because it does not strike at random — it systematically hits the most recently created entities. Which is precisely the population a regulator will ask about.

Set aside, never delete

The fix is not a better join. It is a different contract: the chain may move a row, never delete one.

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())

The outer join keeps everything. The split is explicit. And the row carries the reason it was set aside, which is what makes it recoverable: once the reference data catches up, a scheduled job replays every A_TRAITER row and closes it.

One detail that matters in Spark: an outer join is more expensive than an inner one, and the optimiser can no longer push certain filters through it. At scale, the form that holds is to compute the split once by marking the rows, then derive both outputs from the same cached DataFrame — otherwise the join is evaluated twice.

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

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

The check that makes the loss impossible

One assertion, at the end of every run, makes this entire class of defect impossible:

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

require(
  collecte == declare + ecartees,
  s"silent loss: ${collecte - declare - ecartees} rows"
)

It costs three counts per run. Its value is that it fails the job rather than the filing: a chain that stops at 01:12 is an incident; a chain that files an incomplete return is a finding. Those two do not carry the same price.

It also catches what unit tests do not. Each function was correct on its own; the defect lived in the joint between two of them. Balance checks are the only tests that look at the joints — which is why they must run in production, not only in integration.

Three classic pitfalls with this assertion. It must count the rows received, not the rows read after a dropDuplicates. It must run after the last write, not before. And it must read its counters from the written tables, not from in-memory DataFrames — otherwise it validates your intent, not your result.

An exception is a business object

The last step is the one usually skipped. A quarantine nothing ever leaves is a bin with a better name. Rows leave it only if someone can see them, understand why they are there, and act — which means a screen, not a log file.

• One rule code per row. Aggregate the reasons and you lose the ability to fix the cause.
• An age alert on the backlog. A row pending for three weeks is a stopped process, not a row that is waiting.
• The exception count on the business dashboard. Not in the engineering logs. It is the number that says whether the reference data is keeping up.
• An exit rate, not only a stock. How many rows entered quarantine this week, how many left. If the second number is zero for a month, the correction loop exists only on paper.

If you take one thing from this: go and read the joins in your quality layer, and count how many are inner joins. Then ask what happens to the rows they remove.