Fictional practice · Reference code executed locally · The supplied local reference was executed against the included fixture. This does not verify a production service or the learner’s own environment.
Choose ETL or ELT from the workload
Choose ETL or ELT from the workload
ETL transforms before loading a target; ELT loads before transformations in a capable target platform. Neither removes the need for validation, access control or provenance. Define raw, prepared and serving responsibilities before choosing tools. Haven’s local workflow reads immutable CSVs, validates relationships, derives line values and publishes analytical outputs. It is a snapshot exercise, not an incremental ingestion service. The earlier processing course explains correction evidence; this track adds operating boundaries and stage contracts.
Follow a concrete example
Follow the contract from48 source lines to48 technically accepted records, then42 completed analytical lines. A cancelled status is not a parsing defect. Preserve that distinction in filenames and counts so downstream readers can separate quality exclusions from reporting scope. The manifest should identify source hashes and rule/version rather than merely say success.
Further reading: Airflow 2.10.5 fundamentals, Spark SQL guide. The case and exercises are authored for DataCareerHub.
Your task
Draw an ETL and an ELT version of the same workflow. State where source validation and business transformations occur.
Use learner_worksheet.txt and the provided source/code files. Save your own artifact and compare the separate solution after attempting the task.
Worked solution
Both designs preserve raw evidence. ETL prepares before the analytical target; ELT loads raw/staged data then transforms within the target. Technical architecture does not decide whether a source value is authoritative.
Implement ingestion, transformation and validation
Implement ingestion, transformation and validation
Read source rows with explicit schemas and check parent uniqueness before matching. Parse quantities and prices, retain row-level failures with reason codes and stop on ambiguous relationships. The reference validates all-status lines, then defines completed analytical scope separately. This distinction prevents cancelled lines from being mislabelled as invalid records. Derived value remains integer cents. Its manifest records source hashes, counts and totals; a downstream reader must know which stage produced each artifact.
Follow a concrete example
For a row with an invalid quantity, keep its source-row identity and a reason code. For duplicated order parents, stop before matching because a single authoritative parent cannot be selected. These dispositions are contract decisions, not generic exception handling. Reconcile accepted plus rejected records to the original source count after every mutation.
Inspect the reference implementation
Language: Python 3.12. The complete downloadable file includes the named helpers and the fresh-output command-line runner. Trace the source, transformation and result against this lesson’s contract before running it locally.
def de_pipeline(source,out):
# Full snapshot with staged output; no append or inferred incremental semantics.
from datetime import date
order_rows=rows(source,'orders.csv');line_rows=rows(source,'order_lines.csv');products=rows(source,'products.csv')
if len({r['order_id'] for r in order_rows})!=len(order_rows):raise ValueError('Duplicate order keys')
if len({r['product_id'] for r in products})!=len(products):raise ValueError('Duplicate product keys')
for r in order_rows:
if not r['order_id'] or r['status'] not in ['completed','cancelled']:raise ValueError('Invalid order key/status')
if not re.fullmatch(r'\d{4}-\d{2}-\d{2}',r['order_date']):raise ValueError('ISO calendar date required')
date.fromisoformat(r['order_date'])
orders={r['order_id']:r for r in order_rows};product_keys={r['product_id'] for r in products};accepted=[];rejected=[]
for n,r in enumerate(line_rows,2):
reasons=[]
try:
quantity=int(r['quantity']);price=int(r['unit_price_cents'])
if quantity<=0 or price<0:reasons.append('invalid_domain')
except ValueError:reasons.append('invalid_number');quantity=price=0
if r['order_id'] not in orders:reasons.append('unmatched_order')
if r['product_id'] not in product_keys:reasons.append('unmatched_product')
if not r['line_id']:reasons.append('missing_line_id')
if reasons:rejected.append({'row':n,'line_id':r['line_id'],'reasons':reasons});continue
accepted.append(dict(r,value_cents=quantity*price,status=orders[r['order_id']]['status']))
if len({r['line_id'] for r in accepted})!=len(accepted):raise ValueError('Duplicate line keys')
completed=[r for r in accepted if r['status']=='completed']
manifest={'source_lines':len(line_rows),'accepted_lines_all_statuses':len(accepted),'quarantined_lines':len(rejected),'completed_lines':len(completed),'completed_value_cents':sum(r['value_cents'] for r in completed),'source_hashes':{n:hashlib.sha256((Path(source)/n).read_bytes()).hexdigest() for n in ['orders.csv','order_lines.csv','products.csv']}}
assert len(line_rows)==len(accepted)+len(rejected)
(Path(out)/'prepared.json').write_text(json.dumps(accepted,indent=2)+'\n');(Path(out)/'rejects.json').write_text(json.dumps(rejected,indent=2)+'\n')
return manifest
Further reading: Airflow 2.10.5 fundamentals, Spark SQL guide. The case and exercises are authored for DataCareerHub.
Your task
Run the local pipeline and inspect prepared.json, rejects.json and result.json. Identify valid cancelled lines versus rejected rows.
Use learner_worksheet.txt and the provided source/code files. Save your own artifact and compare the separate solution after attempting the task.
Worked solution
All 48 fixture lines pass technical validation; 42 belong to completed orders and sum to 119000 cents. Six cancelled-order lines are valid source records outside reporting scope, not quality rejections.
Define publication and failure boundaries
Define publication and failure boundaries
A directory can exist even when a run fails partway through file writing. Treat the final manifest as a completion artifact and verify every named file before consuming it. The teaching runner uses a fresh output directory and writes result.json last; it does not claim atomic multi-file publication. A production design can stage, validate and atomically rename within a filesystem or use a transactional target. Separate transient extraction failures from deterministic invalid-data failures when deciding retries.
Follow a concrete example
Interrupt a hypothetical run between writing prepared.json and the final result manifest. A consumer must not accept the incomplete directory. In a real system, stage and atomically publish or use transactions under a documented recovery protocol. The teaching runner’s fresh-output check prevents overwrites but does not claim to make all file writes atomic.
Further reading: Airflow 2.10.5 fundamentals, Spark SQL guide. The case and exercises are authored for DataCareerHub.
Your task
Document a publication contract and test a duplicate order key in a separate copy. Explain what a partial output folder means.
Use learner_worksheet.txt and the provided source/code files. Save your own artifact and compare the separate solution after attempting the task.
Worked solution
A duplicate order key raises an error and must not produce accepted downstream data. A partial directory without a valid final manifest is not a successful run. Atomic publication and concurrent-run control remain explicit production design tasks.