Reverse ETL

In the previous five lessons the data always flowed in one direction: from the source system to the pipeline, from the pipeline to the summary, from the summary to the serving. This direction was the silent assumption of every measure. Yet a number in the serving layer often does not stay there. The expected consumption per region, the risk class of a subscriber, or the threshold of a region does its real work not on a dashboard but in the field: the application that takes the readings uses it, billing applies it, the alerting mechanism looks at it. Writing processed data back to an operational system is called reverse ETL, and it is one of the most legitimate outputs of a pipeline.

Being legitimate does not mean being harmless. If the written-back value changes the behavior of the operational system, the record that system produces changes too; and that record is the pipeline’s input. At this point the arrow closes into a circle. This lesson measures the circle with two numbers: after how many stages it closes, and what the next run gives after it has closed.

• DI57. The code is fictional; no operational system, cloud service, vendor, or orchestrator is called. The source is the fictional meter readings of the Data Engineering and MLOps curriculum: 200 subscribers, 12 periods, 2400 rows, seed 20260218.
• DI58. Reverse ETL is modeled with a single value: the expected consumption threshold per region. It is computed in the serving layer, written to the operational system, and its factor is 1.2.
• DI59. What the operational system does with that threshold is fictional but ordinary: the field application does not record an increment above the threshold, adds the threshold to the previous index, and leaves the reading to be verified. This is the value that gets recorded.
• DI60. The chain is written as a graph. The graph algorithm is not repeated; it was measured in the Graphs topic of the Data Structures course. Here the chain is only walked and its length is counted.
• DI61. The source itself does not change between runs. The only thing that changes is the index that the operational system records; the real meter readings are fixed.
• DI62. The check thresholds are the same as before: 0.05 for the row count tolerance and 0.25 for the distribution tolerance.
• DI63. The units counted are the stages in the cycle, clipped readings, consumption rows, polluted summary cells, the serving deviation, and the difference between runs. No duration is written.
• DI64. The rule of the course is tested directly in this lesson: the number of a pipeline is that it gives the same result when it is run again.

The Length of the Circle

The block below sets up the common definition and adds three things to it: cycle, which walks the chain, ingest, which produces the record of the operational system, and write_back, which writes the threshold from the serving layer. Then it runs the pipeline five times; each run starts with the threshold written by the previous one.

# FICTIONAL. No real operational system , cloud service or orchestrator
# is called; the meter readings of the measurement network are fictional , seed 20260218.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


REGION = ["east", "west", "north", "south", "central"]
TYPE = ["residential", "commercial"]
PERIODS = [f"2026-{a:02d}" for a in range(1, 13)]
SUBSCRIBERS = {}
for i in range(200):
    r = rng(SEED + 37 * i)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "start": 10000 + int(r() * 60000)}


def generate_source():                              # one reading per subscriber per period
    records = []
    for i, (sub, a) in enumerate(SUBSCRIBERS.items()):
        e = a["start"]
        for p in PERIODS:
            r = rng(SEED + 101 * i + 7 * PERIODS.index(p))
            e += 4 + int(r() * 46)
            records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"],
                            "index": e, "unit": "m3"})
    return records


def s1_screen(records):
    REQUIRED = ("subscriber", "period", "region", "index", "unit")
    keep = [k for k in records if all(a in k for a in REQUIRED)]
    return keep, len(records) - len(keep)


def s2_dedupe(records):
    seen, keep = set(), []
    for k in records:
        key = (k["subscriber"], k["period"])
        if key in seen:
            continue
        seen.add(key)
        keep.append(k)
    return keep, len(records) - len(keep)


def s3_consumption(records):                        # consumption from the period difference
    table = {(k["subscriber"], k["period"]): k for k in records}
    out, dropped = [], 0
    for k in records:
        i = PERIODS.index(k["period"])
        if i == 0:
            continue
        prev = table.get((k["subscriber"], PERIODS[i - 1]))
        if prev is None:
            dropped += 1
            continue
        out.append(dict(k, m3=int(k["index"]) - int(prev["index"])))
    return out, dropped


def s4_summary(records):
    totals = {}
    for k in records:
        key = (k["region"], k["period"])
        totals[key] = totals.get(key, 0) + k["m3"]
    return totals, 0


def s5_serve(summary):                              # per-region mean over the periods
    region = {}
    for (b, d), v in summary.items():
        region.setdefault(b, []).append(v)
    return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0


CHAIN = {"ingest": "screen", "screen": "dedupe", "dedupe": "consumption",
         "consumption": "summary", "summary": "serve", "serve": "write_back", "write_back": "ingest"}


def cycle(start):
    """Walks the chain and collects the stages passed until it returns to the start."""
    path, nxt = [start], CHAIN[start]
    while nxt != start:
        path.append(nxt)
        nxt = CHAIN[nxt]
    return path


def ingest(raw, threshold):
    """FICTIONAL operational system: does not record an increment above the threshold , adds the threshold instead."""
    records, last, clipped = [], {}, 0
    for k in raw:
        e, prev, s = k["index"], last.get(k["subscriber"]), threshold.get(k["region"])
        if prev is not None and s is not None and e - prev > s:
            e = prev + s
            clipped += 1
        last[k["subscriber"]] = e
        records.append(dict(k, index=e))
    return records, clipped


def pipeline(records):
    v, _ = s1_screen(records)
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    summ, _ = s4_summary(v)
    serving, _ = s5_serve(summ)
    return v, summ, serving


def write_back(v, factor=1.2):
    """Reverse ETL: the expected consumption threshold per region is written to the operational system."""
    totals, count = {}, {}
    for k in v:
        totals[k["region"]] = totals.get(k["region"], 0) + k["m3"]
        count[k["region"]] = count.get(k["region"], 0) + 1
    return {b: int(round(factor * totals[b] / count[b])) for b in totals}


raw = generate_source()
path = cycle("ingest")
_, BASELINE, BASE_SERVING = pipeline(ingest(raw, {})[0])
print("cycle:", " -> ".join(path), "-> ingest")
print("stages in the cycle", len(path), ", baseline consumption 2200 , summary cells", len(BASELINE))
print()
print("  run   clipped   east thr   central    north     east     west    south  vs previous")
threshold, prev = {}, None
for i in range(1, 6):
    records, clip = ingest(raw, threshold)
    v, summ, serving = pipeline(records)
    f = "-" if prev is None else max(abs(round(serving[b] - prev[b], 2)) for b in serving)
    print(f"{i:>5} {clip:>9} {threshold.get('east', 0):>10}  " +
          " ".join(f"{serving[b]:>8}" for b in ("central", "north", "east", "west", "south")) +
          f" {f:>12}")
    prev, threshold = serving, write_back(v)

cycle: ingest -> screen -> dedupe -> consumption -> summary -> serve -> write_back -> ingest
stages in the cycle 7 , baseline consumption 2200 , summary cells 55

  run   clipped   east thr   central    north     east     west    south  vs previous
    1         0          0   1321.45  1212.55   881.36   833.09  1065.55            -
    2      1129         31   1274.45  1177.73   853.45   813.82  1033.09         47.0
    3      1207         30   1263.55  1167.45   845.55   813.82  1025.36         10.9
    4      1225         30   1263.55  1153.09   845.55   813.82  1025.36        14.36
    5      1225         30   1263.55  1153.09   845.55   813.82  1025.36          0.0

After Seven Stages, Its Own Input

The first line gives the circle. The chain leaves the ingest node, passes the five stages of the pipeline, enters the reverse ETL write, and from there returns to ingest again: seven stages. The number itself is not a warning; what is a warning is that the chain is now a graph cycle. Before reverse ETL was added, the chain was six nodes and five edges, with a clear start and end. The moment the seventh edge is added, no node is “run first” any more; every node has a predecessor.

In practice this means that the pipeline no longer has a definition independent of the run. The answer to the question “what is the output of the pipeline” is not a number but a run number. The last column of the table counts this: the second run deviates from the first by 47.00, the third from the second by 10.90, the fourth from the third by 14.36. Run with the same source, the same code, and the same seed, the pipeline gives three different answers.

The mechanism can be read in the middle columns. The first run goes without a threshold and gives the right answer; at the end it writes a threshold of 31 for the east region to the operational system. In the second run the field application uses this threshold and clips 1129 readings: increments above the threshold are not recorded. The mean computed from the clipped readings falls, the lower mean produces a smaller threshold, and the smaller threshold clips 1207 readings in the third run. The loop feeds itself and narrows the input a little more with every turn.

Pay attention to where the seventh edge sits: not in the code of the pipeline. The pipeline only writes a value and ends there; the clipping decision is inside the operational system. The circle is built between two systems and appears in the code of neither. Someone who looks at the graph of the pipeline sees a neat chain of six nodes; someone who looks at the field application sees a reasonable validation rule. Only someone who draws the two at the same time can see the circle.

Where the Loop Stops

The fifth row says that the loop has stopped: the clipped readings settle at 1225 and the difference between runs becomes 0.00. After the fourth run, when the pipeline is run again, it gives the same result. The block below measures what the settled run is and runs the same pipeline in a setup where the write-back does not touch the input.

records, clip = ingest(raw, threshold)
v, summ, serving = pipeline(records)
print("settled run: consumption", len(v), ", dirty cells",
      sum(1 for a in set(summ) | set(BASELINE) if summ.get(a) != BASELINE.get(a)))
print("serving deviation from baseline", {b: round(serving[b] - BASE_SERVING[b], 2) for b in BASE_SERVING})
print("largest summary cell deviation",
      round(max(abs(summ[a] - BASELINE[a]) / abs(BASELINE[a]) for a in BASELINE), 4), "< threshold 0.25")

print()
print("the written-back threshold does not touch the input:")
print("  run   clipped   central    north     east     west    south  vs previous")
threshold, prev = {}, None
for i in range(1, 4):
    records, clip = ingest(raw, {})
    v, summ, serving = pipeline(records)
    f = "-" if prev is None else max(abs(round(serving[b] - prev[b], 2)) for b in serving)
    print(f"{i:>5} {clip:>9}  " +
          " ".join(f"{serving[b]:>8}" for b in ("central", "north", "east", "west", "south")) +
          f" {f:>12}")
    prev, threshold = serving, write_back(v)
print("threshold written to the operational system", threshold)

settled run: consumption 2200 , dirty cells 55
serving deviation from baseline {'central': -57.9, 'north': -59.46, 'east': -35.81, 'west': -19.27, 'south': -40.19}
largest summary cell deviation 0.1788 < threshold 0.25

the written-back threshold does not touch the input:
  run   clipped   central    north     east     west    south  vs previous
    1         0   1321.45  1212.55   881.36   833.09  1065.55            -
    2         0   1321.45  1212.55   881.36   833.09  1065.55          0.0
    3         0   1321.45  1212.55   881.36   833.09  1065.55          0.0
threshold written to the operational system {'central': 32, 'north': 32, 'east': 31, 'west': 32, 'south': 32}

The first three lines say what the settled run is, and the answer is disturbing. The consumption rows are 2200, exactly the baseline value; no row has been dropped and none has multiplied. Yet 55 of the 55 summary cells are dirty: every period of every region has been computed wrongly. The serving deviates between −19.27 and −59.46. And no check fires: the row count is complete, the schema is complete, the keys are unique, every region is present in every period, and the largest cell deviation of 0.1788 stays below the 0.25 tolerance.

This table also corrects how the rule of the course should be read. After the fourth run, when the pipeline is run again, it gives the same result, so the rule is satisfied — but the result is wrong. The rule is a necessary condition, not a sufficient one: a pipeline that cannot be re-run has not been measured, while a pipeline that can be re-run has been measured only if it does not re-read its own output. In a looped pipeline, stability is a sign not of correctness but of convergence; the pipeline has come to rest at a constraint that it produced itself.

The only check that could see this corruption is already in the last column of the first table: a check that compares the serving value of the same period with that of the previous run. That check would fire in the second, third, and fourth runs, because it would see differences of 47.00, 10.90, and 14.36 respectively. After the fifth run it would go silent. That is, comparing runs catches a loop only before it settles; after it has settled, all the indicators of the pipeline are green and the served value is permanently wrong. A loop is the only form of corruption whose catch window closes by itself.

The second table shows the solution, and the solution is not to give up reverse ETL. The threshold is still computed and still written to the operational system — central 32, north 32, east 31, west 32, south 32 — but the field application uses it only to call the verification, not to change the record. The raw reading is stored as it is. The clipped readings are 0, the difference between runs is 0.00, and all three runs give the baseline. Reverse ETL is an output; the moment it becomes the pipeline’s input it ceases to be an output. What makes the distinction is not whether the value is written, but whether the written-back field is the field the pipeline reads.

Summary

• Reverse ETL is writing processed data back to an operational system, and if the written-back value affects the record that system produces, the chain of the pipeline closes into a circle: the path leaving the ingest node returns to its own input after seven stages.
• The threshold written by the first run clips 1129 readings in the second run, 1207 in the third and 1225 in the fourth; with the same source and the same code, the difference between runs comes out as 47.00, 10.90, and 14.36.
• The loop stops at the fourth run but the stopping point is not the baseline: while the consumption rows stay at 2200, 55 of the 55 summary cells are polluted and the serving deviates by −19.27 … −59.46.
• No check catches this corruption; the row count, schema, uniqueness, and partition completeness are all intact, and the largest cell deviation of 0.1788 stays below the 0.25 tolerance.
• A check that compares runs would see the loop in the second, third, and fourth runs and would go silent after the fifth. The catch window of a loop closes by itself.
• When the same threshold is computed and written but the raw record is not touched, the clipped readings are 0 and the difference between runs is 0.00: what decides is not that the value is written, but whether the pipeline reads its own output.

Next Step

In six lessons the data was ingested into the pipeline. The rows the ingestion method carried and the changes it missed were counted, the call completed under the constraint of the source was measured, the partition a message would come from and what the number of partitions does to ordering were written down, the result a window decision changes for a late-arriving record and the cost of carrying state were derived, the loss and duplication of the three delivery modes were counted separately, and the circle that the write-back opens was measured. What they all have in common is this: the row entered the pipeline. What was done to it after it entered was never asked. In all of these lessons the five stages of the pipeline were called like a ready-made box; where which transformation would be done, whether the same computation would run at the source or at the target, how far down the layers the raw row would be kept, and how to write whether a transformation is correct were not defined. The next topic does exactly that: it chooses the place of the transformation, separates the layers, and counts which corruption each layer can stop.