--- title: 'The Data Warehouse and Transformation Pipelines' source: 'https://academia.sh/en/courses/enterprise-context/data-warehouse-and-transformation-pipelines' course: 'Enterprise Context and Integration' language: en updated: '2026-08-23T07:01:06+00:00' license: 'CC BY-SA 4.0' --- # The Data Warehouse and Transformation Pipelines Separating analytical load from operational load: five reports imposing 163 queries and 5,354,308 records read per day on the operational system, the warehouse taking on that load in exchange for a freshness lag, an eight-step transformation pipeline moving 523,341 records in one cycle, and a single field change in the source schema breaking one step and stopping four more, dropping all five of the five reports. The previous lesson dealt with a case where multiple systems wrote the same entity: a source system was chosen and the inconsistency between records was counted. There, every party was **writing** the data. In this lesson there is a party that never writes at all. The regional library network's management unit only asks questions: which branch is filling up, which item is overdue, how membership is flowing. A question does not change the data, but it is not free either; every question becomes a query, every query becomes records read, and those records sit on the disk of the system that runs loan transactions. This lesson measures the separation of analytical load from operational load. The network is a **fictional model** whose owners and budgets are separate. ## The Questioning Party Brings a Countable Load The model has four sources, each with a separate owner: the loan service, the membership system, the externally sourced catalog, and the branch systems. The management unit's five questions read these. **IN10: the record counts, run frequencies, and freshness requirements are fictional** and are not taken from any real institution. **IN11: a report touching one entity is one query; the event entity is read by window, the dimension entity is scanned in full.** **IN14: a step that cannot find the field it reads breaks; a step whose input is not intact stops.** ```js // warehouse/model.mjs — the regional library network's fictional model and transformation pipeline. // Records are actually generated; the numbers in the next blocks are measured by scanning these arrays. export const SEED = 7391, DAYS = 180, END = DAYS * 24 - 1; export const OWNER = { loan: "loan service (in-house)", member: "membership system", item: "catalog (external)", branch: "branch systems" }; export function generate(t = SEED) { let s = t; const r = () => (s = (s * 48271) % 2147483647) / 2147483647, n = (k) => Math.floor(r() * k); const array = (k, f) => Array.from({ length: k }, (_, i) => f(i)); return { branch: array(12, (i) => ({ id: i, region: `b${i % 4}` })), item: array(128_000, (i) => ({ id: i, type: ["book", "periodical", "audio"][n(3)], changeStamp: n(END) })), member: array(46_000, (i) => ({ id: i, regDay: n(DAYS), branchId: n(12), changeStamp: n(END) })), loan: array(240_000, (i) => ({ id: i, memberId: n(46_000), itemId: n(128_000), branchId: r() < 0.008 ? null : n(12), day: n(DAYS), hour: n(24) })), }; } // Reports: entity -> window (days) or full scan; runs per day; largest accepted lag (hours); // basis in the warehouse; field that determines breakage. export const REPORTS = { occupancy: { reads: { loan: 1, branch: "full" }, runsPerDay: 48, freshness: 1, basis: "branch-summary", field: "loan.branchId" }, "overdue-item": { reads: { loan: 30, item: "full" }, runsPerDay: 24, freshness: 4, basis: "enriched", field: "item.type" }, "new-member": { reads: { member: "full" }, runsPerDay: 12, freshness: 14, basis: "member-summary", field: "member.regDay" }, activity: { reads: { loan: 90, member: "full" }, runsPerDay: 2, freshness: 36, basis: "member-summary", field: "member.regDay" }, turnover: { reads: { loan: 180, item: "full", branch: "full" }, runsPerDay: 1, freshness: 72, basis: "enriched", field: "item.type" }, }; // Records read and queries opened in a single run. export function readReport(data, r) { let recordsRead = 0, queries = 0; for (const [entity, window] of Object.entries(r.reads)) { queries += 1; if (window === "full") { recordsRead += data[entity].length; continue; } for (const k of data[entity]) if (k.day >= DAYS - window) recordsRead += 1; } return { recordsRead, queries }; } // The slice pulled in one cycle: from the event records within the period, from the dimension // either everything or records whose change stamp falls within the period (IN13). export function extract(data, period, dimension) { const threshold = END - period; const changed = (d) => dimension === "full" ? d : d.filter((k) => k.changeStamp > threshold); const slice = { loan: data.loan.filter((k) => k.day * 24 + k.hour > threshold), member: changed(data.member), item: changed(data.item), branch: data.branch }; return { slice, recordsRead: Object.values(slice).reduce((t, d) => t + d.length, 0), queries: 4 }; } // The pipeline's eight steps; each step declares the field it reads. export const STEPS = [ { name: "extract-loan", source: "loan", inputs: [], reads: ["memberId", "itemId", "branchId", "day"] }, { name: "extract-member", source: "member", inputs: [], reads: ["regDay", "branchId"] }, { name: "extract-item", source: "item", inputs: [], reads: ["type"] }, { name: "extract-branch", source: "branch", inputs: [], reads: ["region"] }, { name: "clean", inputs: ["extract-loan"], reads: ["branchId", "day"] }, { name: "enriched", inputs: ["clean", "extract-item", "extract-branch"], reads: ["itemId", "type", "region"] }, { name: "branch-summary", inputs: ["enriched"], reads: ["region", "day"] }, { name: "member-summary", inputs: ["clean", "extract-member"], reads: ["memberId", "regDay"] }, ]; const group = (d, f) => [...new Set(d.map(f))].map((a) => ({ a })); const TRANSFORM = { clean: ([l]) => l.filter((k) => k.branchId !== null), enriched: ([l, i, b]) => { const t = new Map(i.map((k) => [k.id, k.type])), r = new Map(b.map((k) => [k.id, k.region])); return l.map((k) => ({ ...k, type: t.get(k.itemId) ?? "unknown", region: r.get(k.branchId) ?? "unknown" })); }, "branch-summary": ([l]) => group(l, (k) => `${k.region}|${k.day}`), "member-summary": ([l]) => group(l, (k) => `${k.memberId}|${Math.floor(k.day / 30)}`), }; // Run the pipeline: a step whose transform throws is "broken", a step whose input is not intact is "stopped". export function run(slice) { const output = new Map(), status = new Map(), counts = new Map(); let moved = 0; for (const s of STEPS) { if (s.inputs.some((d) => status.get(d) !== "done")) { status.set(s.name, "stopped"); continue; } const feed = s.source ? [slice[s.source]] : s.inputs.map((d) => output.get(d)); const sample = feed.filter((d) => d.length > 0).map((d) => d[0]); try { for (const field of s.reads) if (sample.every((k) => (field in k) === false)) throw new Error(`missing field ${field}`); const result = s.source ? feed[0] : TRANSFORM[s.name](feed); const inCount = feed.reduce((t, d) => t + d.length, 0); output.set(s.name, result); status.set(s.name, "done"); counts.set(s.name, { inCount, outCount: result.length }); moved += inCount + result.length; } catch (e) { status.set(s.name, `broken: ${e.message}`); } } return { status, counts, moved }; } ``` ## The Warehouse Takes On the Load, in Exchange for a Lag In one cycle the warehouse reads the sources, transforms them, and leaves behind the summaries the reports will read. A report gets its answer from the warehouse only if the lag it accepts is larger than the warehouse's own lag. **IN12: the pipeline moves 4,000 records per second; the freshness lag is the sum of the period and the run duration.** **IN13: incremental dimension loading assumes an index on the change stamp.** ```js // warehouse/load.mjs — queries and records read that land on the operational system: without a warehouse and with one import { generate, REPORTS, readReport, extract, run, DAYS } from "./model.mjs"; const data = generate(); const SPEED = 4000; // IN12: the pipeline moves 4,000 records per second const measurements = Object.fromEntries(Object.entries(REPORTS).map(([name, r]) => [name, readReport(data, r)])); console.log(`model: ${Object.entries(data).map(([name, d]) => `${name} ${d.length}`).join(", ")}; ${DAYS} days`); console.log("\nreport runs/day queries/run records/run records/day freshness (hrs)"); console.log("-------------- ---------- ------------ ------------- ----------- ---------------"); let rawRead = 0, rawQueries = 0; for (const [name, r] of Object.entries(REPORTS)) { const o = measurements[name]; rawRead += o.recordsRead * r.runsPerDay; rawQueries += o.queries * r.runsPerDay; console.log(`${name.padEnd(14)} ${String(r.runsPerDay).padStart(10)} ${String(o.queries).padStart(12)} ` + `${String(o.recordsRead).padStart(13)} ${String(o.recordsRead * r.runsPerDay).padStart(11)} ${String(r.freshness).padStart(15)}`); } console.log(`without warehouse, total: ${rawQueries} queries/day, ${rawRead} records read/day`); console.log("\nperiod dimension run (s) lag in whs direct queries/day records/day % of raw"); console.log("------ ------------ --------- ------ ------- ------- ------------ ----------- ----------"); const remainingByPeriod = {}; for (const dimension of ["full", "incremental"]) for (const period of [24, 12, 3, 1]) { const c = extract(data, period, dimension); const duration = run(c.slice).moved / SPEED; const lag = period + duration / 3600, cycles = 24 / period; let queries = c.queries * cycles, recordsRead = c.recordsRead * cycles; const remaining = []; for (const [name, r] of Object.entries(REPORTS)) { if (r.freshness >= lag) continue; remaining.push(name); queries += measurements[name].queries * r.runsPerDay; recordsRead += measurements[name].recordsRead * r.runsPerDay; } remainingByPeriod[period] = remaining; console.log(`${String(period).padStart(6)} ${dimension.padEnd(12)} ${duration.toFixed(1).padStart(9)} ` + `${lag.toFixed(2).padStart(6)} ${String(5 - remaining.length).padStart(7)} ${String(remaining.length).padStart(7)} ` + `${String(Math.round(queries)).padStart(12)} ${String(Math.round(recordsRead)).padStart(11)} ` + `${`${(recordsRead / rawRead * 100).toFixed(1)}%`.padStart(10)}`); } console.log(`\nreports still operational at the 24-hour period: ${remainingByPeriod[24].join(", ")}`); console.log(`reports still operational at the 3-hour period: ${remainingByPeriod[3].join(", ")}`); ``` ``` model: branch 12, item 128000, member 46000, loan 240000; 180 days report runs/day queries/run records/run records/day freshness (hrs) -------------- ---------- ------------ ------------- ----------- --------------- occupancy 48 2 1326 63648 1 overdue-item 24 2 168293 4039032 4 new-member 12 1 46000 552000 14 activity 2 2 165808 331616 36 turnover 1 3 368012 368012 72 without warehouse, total: 163 queries/day, 5354308 records read/day period dimension run (s) lag in whs direct queries/day records/day % of raw ------ ------------ --------- ------ ------- ------- ------------ ----------- ---------- 24 full 133.4 24.04 2 3 160 4830006 90.2% 12 full 131.9 12.04 3 2 152 4451992 83.1% 3 full 130.8 3.04 4 1 128 1456912 27.2% 1 full 130.6 1.04 4 1 192 4241160 79.2% 24 incremental 3.6 24.00 2 3 160 4656902 87.0% 12 incremental 1.8 12.00 3 2 152 4104854 76.7% 3 incremental 0.4 3.00 4 1 128 65392 1.2% 1 incremental 0.1 1.00 4 1 192 65160 1.2% reports still operational at the 24-hour period: occupancy, overdue-item, new-member reports still operational at the 3-hour period: occupancy ``` The first table shows where the load sits. There are 163 queries and 5,354,308 records read per day; 4,039,032 of that, or seventy-five percent, comes from **a single report**. The reason is not the number of runs but the records read per run: the overdue-item list scans the catalog's 128,000 rows on every run. The occupancy report, which runs 48 times a day, does not even reach one percent of the total. The second table counts the share the warehouse takes on, period by period, and it draws a two-way curve. As the period shortens, the lag drops, and as the lag drops, more reports can get their answer from the warehouse: from two to four. But the cycle count also rises, and every cycle re-reads the sources. With full dimension loading, records read come to 90.2 percent at the 24-hour period, 27.2 percent at the 3-hour period, and back up to 79.2 percent at the 1-hour period. The cheapest point is not at either end, it is in the middle. The column that actually decides is **dimension**. At the same period, incremental loading brings 27.2 percent down to 1.2 percent, because the catalog's 128,000 rows are read only for their changed portion, not on every cycle. Whether the operational load drops is not decided by the warehouse's existence but by how it reads the sources. One report never moves to the warehouse at any period: the occupancy report accepts a lag of one hour, and the shortest period's lag is 1.04 hours. The freshness requirement defines the path the warehouse cannot cover. ## The Pipeline's Steps and a Change in the Source Schema The pipeline is eight steps, and each step declares the field it reads. A field change in the source schema is actually applied: the field is renamed or removed, the pipeline is run, and the broken step is counted. ```js // warehouse/breakage.mjs — records moved step by step, and the step a source-schema change breaks import { generate, REPORTS, extract, run, STEPS, OWNER } from "./model.mjs"; const data = generate(), PERIOD = 3; const { slice } = extract(data, PERIOD, "full"); const baseline = run(slice); const lag = PERIOD + baseline.moved / 4000 / 3600; console.log("step input in out"); console.log("--------------- ----------------------------------- -------- --------"); for (const s of STEPS) { const c = baseline.counts.get(s.name); console.log(`${s.name.padEnd(15)} ${(s.source ?? s.inputs.join(",")).padEnd(35)} ` + `${String(c.inCount).padStart(8)} ${String(c.outCount).padStart(8)}`); } console.log(`one cycle moved ${STEPS.length} steps, ${baseline.moved} records; lag ${lag.toFixed(2)} hours`); // Three schema changes applied: a field is renamed or removed. const CHANGES = { "item.type -> item.typeCode": { entity: "item", field: "type", newName: "typeCode" }, "loan.branchId -> loan.branch": { entity: "loan", field: "branchId", newName: "branch" }, "member.regDay removed": { entity: "member", field: "regDay", newName: null }, }; const apply = (d, c) => ({ ...d, [c.entity]: d[c.entity].map((k) => { const y = { ...k }; delete y[c.field]; if (c.newName) y[c.newName] = k[c.field]; return y; }) }); console.log("\nchange owner broken stopped intact dropped in whs broken w/o whs"); console.log("----------------------------- --------------------------- ------ ------- ------- -------------- --------------"); for (const [name, c] of Object.entries(CHANGES)) { const d = [...run(apply(slice, c)).status]; const count = (f) => d.filter(([, v]) => f(v)).length; const broken = new Set(d.filter(([, v]) => v !== "done").map(([k]) => k)); const dropped = Object.values(REPORTS).filter((r) => broken.has(r.basis)).length; const direct = Object.values(REPORTS).filter((r) => r.field === `${c.entity}.${c.field}`).length; console.log(`${name.padEnd(29)} ${OWNER[c.entity].padEnd(27)} ${String(count((v) => v.startsWith("broken"))).padStart(6)} ` + `${String(count((v) => v === "stopped")).padStart(7)} ${String(count((v) => v === "done")).padStart(7)} ` + `${String(dropped).padStart(14)} ${String(direct).padStart(14)}`); } // Edge: without a warehouse, a report connects to every source it reads; with one, only the pipeline does. const rawEdges = Object.values(REPORTS).flatMap((r) => Object.keys(r.reads)); const uncovered = Object.values(REPORTS).filter((r) => r.freshness < lag); const uncoveredEdges = uncovered.flatMap((r) => Object.keys(r.reads)); console.log(`\nedges: without warehouse ${rawEdges.length} (report -> source); with warehouse ${4 + (5 - uncovered.length) + uncoveredEdges.length} ` + `(4 pulls + ${5 - uncovered.length} warehouse reads + ${uncoveredEdges.length} uncovered direct reads)`); console.log(`owners the management unit talks to: without warehouse ${new Set(rawEdges).size}, ` + `with warehouse ${new Set(["warehouse", ...uncoveredEdges]).size}`); ``` ``` step input in out --------------- ----------------------------------- -------- -------- extract-loan loan 146 146 extract-member member 46000 46000 extract-item item 128000 128000 extract-branch branch 12 12 clean extract-loan 146 144 enriched clean,extract-item,extract-branch 128156 144 branch-summary enriched 144 4 member-summary clean,extract-member 46144 143 one cycle moved 8 steps, 523341 records; lag 3.04 hours change owner broken stopped intact dropped in whs broken w/o whs ----------------------------- --------------------------- ------ ------- ------- -------------- -------------- item.type -> item.typeCode catalog (external) 1 2 5 3 2 loan.branchId -> loan.branch loan service (in-house) 1 4 3 5 1 member.regDay removed membership system 1 1 6 2 2 edges: without warehouse 10 (report -> source); with warehouse 10 (4 pulls + 4 warehouse reads + 2 uncovered direct reads) owners the management unit talks to: without warehouse 4, with warehouse 3 ``` The step table shows where the cost comes from. In the three-hour cycle, only 146 records come in from the loan side, but the total moved is 523,341. The whole difference is dimension reading: the `enriched` step takes in 128,156 records, because enriching 144 loan records means reading the entire catalog. The `clean` step passes 144 of the 146 records through; the two records with an empty branch are dropped. The breakage table gives this course's question. All three changes touch a single field, and all three break exactly **one** step of the pipeline. What differs is where the break stops. When the catalog's field breaks, two more steps stop and three reports are left unanswered — without a warehouse, only two reports used that field. When the loan service's field breaks, four steps stop and **all five of the five reports** drop — yet only one report read that field directly. In the membership field, the two counts come out equal. The warehouse ties together reports that read the same source. The occupancy report never uses the catalog's type field; without a warehouse it would be unaffected by that change, but with a warehouse it is affected, because the `branch-summary` step sits downstream of the `enriched` step. The compensation is that fixing it is localized: the break is a single step and belongs to a single team's responsibility. Without a warehouse, the same fix has to be made separately, in each report's independently written query. The last lines count the edges. The edge count does not change: 10 without a warehouse, 10 with one. What changes is direction and ownership. The management unit talks to four owners without a warehouse and three with one; the two remaining owners are there because of the occupancy report the warehouse cannot cover. An integration point does not reduce the edge count — it changes who the coordination is with. ## Summary - Five reports impose 163 queries and 5,354,308 records read per day on the operational system; seventy-five percent of that comes from a single report, because it scans the catalog's 128,000 rows on every run (IN10, IN11). - As the period shortens, the reports answered from the warehouse rise from two to four, but because the cycle count grows, records read climb back from 27.2 percent to 79.2 percent; the cheapest period is not at either end, it is in the middle. - Incremental dimension loading brings the operational load down from 27.2 percent to 1.2 percent at the same period; what decides it is not the warehouse's existence but how it reads the sources (IN13). - The report with a one-hour freshness requirement never moves to the warehouse at any period, because the shortest period's lag is 1.04 hours. This is how the warehouse reports the path it cannot cover (IN12). - The eight-step pipeline moves 523,341 records in one cycle; a single field change in the source breaks a single step, but in the loan field four steps stop downstream of the break, so all five of the five reports drop — only one report read that field directly (IN14). ## Next Step This lesson's measurements leaned on an assumption: that the source systems' fields are known, and that a schema change shows up as an error message. Not every system on the network is like that. There is a record system that has held years of loan history, and no one can reach its source code. The next lesson takes this up: when an unchangeable system's call surface is wrapped, how many paths close, how many stay outside the wrapper, and how long the window lasts where two systems write together while data ownership is handed off.