Ingest Raw Data to Bronze Layer on S3 — Cloud Data Engineering

This exercise builds Stage 1 and Stage 2 of the capstone pipeline: Terraform infrastructure validation and Bronze S3 ingestion. Using only Python and pandas, you will validate the complete Terraform configuration for the IPL lakehouse (CIDR non-overlap, required FinOps tags on all resources, S3 bucket security settings, no public database ports in security groups), implement the idempotent Lambda trigger that activates on scorecard arrival, write raw delivery records to the Bronze S3 path, create an Iceberg-style snapshot, and enforce immutability by blocking overwrites on the Bronze path.

Analogy🏏Cricket

🏏 Think of it like cricket: This exercise is the IPL official statistics team building their daily automated processing pipeline — the complete workflow that takes raw ball-by-ball records from every ground and produces the certified statistics published on the official website by midnight. Stage 1 is the data catalogue check: verify that the incoming scorecards match the expected format before any processing begins. Stage 2 is the statistics calculation: joins with match metadata, derivation of over-level stats, economy rate computation. Stage 3 is the official record update: load the new statistics into the production database using the certified upsert protocol — delete the old version of today's record and insert the freshly computed one — so no match ever has two records in the official database.

The Terraform validation step mirrors the pre-deployment check from Module 1 Exercise and the FinOps tag validation from Module 5. The Lambda trigger implementation re-uses the idempotency pattern from Module 5 Lesson 27. The Bronze landing with Iceberg snapshot connects Module 5 Lesson 25 (Iceberg) with Lesson 26 (Medallion). Completing all assertions in this exercise provides the Bronze foundation layer that the Lesson 33 Silver transformation reads from.

Step 1 — Terraform Validation and Lambda Trigger

Define the complete Terraform configuration for the IPL lakehouse as Python dicts, run all four validation checks (CIDR overlap, required tags on all resources, S3 prevent_destroy and block_public, no database ports open to the internet), implement the idempotent Lambda trigger that logs pipeline runs to a dictionary keyed by source file path, and verify the trigger correctly activates for scorecard CSV files and skips non-scorecard files. Assert trigger idempotency by calling it twice for the same file and confirming only one pipeline run is recorded.

Analogy🏏Cricket

🏏 Think of it like cricket: this step is the match referee certifying the ground and the turnstile system that admits only the official scorecard. Just as the referee confirms no two boundary ropes overlap and every subnet of the outfield sits inside the stadium perimeter, `validate_cidrs` checks CIDR non-overlap and that each subnet is inside the VPC. Just as every asset must carry its full ownership label — Team, Project, Owner and the rest — before the finance officer signs off, `validate_tags` demands all six required tags on every resource, while the S3 security and no-public-database-port checks are the referee confirming the vault is locked and no gate opens straight onto the pitch. The idempotent Lambda trigger is the turnstile that admits one scorecard exactly once: `lambda_trigger` records each source key and, called twice for the same file, starts only a single run while skipping non-scorecard files entirely. The payoff: a fully certified, secure ground whose pipeline fires exactly once per genuine scorecard — no duplicate runs, no exposed entrances.

python

# capstone_bronze_ingestion.py — Step 1: Terraform validation and Lambda trigger
import ipaddress
import uuid
import pandas as pd
import numpy as np
from datetime import datetime, timezone, timedelta
from collections import defaultdict

np.random.seed(42)

# ── Terraform configuration ────────────────────────────────────────────────────
REQUIRED_TAGS = {"Team", "Project", "Environment", "Pipeline", "Owner", "ManagedBy"}
IPL_INFRA = {
    "vpc": {
        "cidr": "10.0.0.0/16",
        "tags": {"Team":"data-eng","Project":"ipl-lakehouse","Environment":"prod",
                 "Pipeline":"capstone","Owner":"[email protected]","ManagedBy":"Terraform"},
    },
    "subnets": [
        {"cidr":"10.0.1.0/24","az":"ap-south-1a","type":"public",
         "tags":{"Team":"data-eng","Project":"ipl-lakehouse","Environment":"prod",
                 "Pipeline":"capstone","Owner":"[email protected]","ManagedBy":"Terraform"}},
        {"cidr":"10.0.10.0/24","az":"ap-south-1a","type":"private",
         "tags":{"Team":"data-eng","Project":"ipl-lakehouse","Environment":"prod",
                 "Pipeline":"capstone","Owner":"[email protected]","ManagedBy":"Terraform"}},
    ],
    "s3_bronze": {
        "bucket":"ipl-lakehouse-bronze-prod",
        "versioning":True,"encryption":"AES256",
        "block_public":True,"prevent_destroy":True,
        "tags":{"Team":"data-eng","Project":"ipl-lakehouse","Environment":"prod",
                "Pipeline":"capstone","Owner":"[email protected]","ManagedBy":"Terraform"},
    },
    "security_groups": [
        {"name":"glue-sg",
         "rules":[
             {"type":"egress","port":-1,"source":"0.0.0.0/0","protocol":"-1"},
         ]},
    ],
    "glue_role": {
        "name":"ipl-glue-execution",
        "trust_service":"glue.amazonaws.com",
        "tags":{"Team":"data-eng","Project":"ipl-lakehouse","Environment":"prod",
                "Pipeline":"capstone","Owner":"[email protected]","ManagedBy":"Terraform"},
    },
}

def validate_cidrs(infra: dict) -> list:
    cidrs = [s["cidr"] for s in infra["subnets"]] + [infra["vpc"]["cidr"]]
    errors = []
    nets = [ipaddress.IPv4Network(c) for c in cidrs[:-1]]
    vpc  = ipaddress.IPv4Network(cidrs[-1])
    for i, n1 in enumerate(nets):
        for n2 in nets[i+1:]:
            if n1.overlaps(n2):
                errors.append(f"CIDR overlap: {n1} ∩ {n2}")
        if not n1.subnet_of(vpc):
            errors.append(f"Subnet {n1} not inside VPC {vpc}")
    return errors

def validate_tags(infra: dict) -> list:
    resources = [infra["vpc"]] + infra["subnets"] + [infra["s3_bronze"], infra["glue_role"]]
    errors = []
    for r in resources:
        missing = REQUIRED_TAGS - set(r.get("tags", {}).keys())
        if missing:
            errors.append(f"Resource '{r.get('name',r.get('bucket',r.get('cidr','?')))}' missing: {missing}")
    return errors

def validate_s3_security(infra: dict) -> list:
    s3, errors = infra["s3_bronze"], []
    if not s3.get("versioning"):     errors.append("S3: versioning not enabled")
    if not s3.get("block_public"):   errors.append("S3: public access not blocked")
    if not s3.get("prevent_destroy"): errors.append("S3: prevent_destroy missing")
    return errors

def validate_sg_no_public_db(infra: dict) -> list:
    DB_PORTS = {5439, 5432, 3306}
    errors = []
    for sg in infra["security_groups"]:
        for r in sg["rules"]:
            if r["type"]=="ingress" and r.get("port") in DB_PORTS and r["source"]=="0.0.0.0/0":
                errors.append(f"SG '{sg['name']}': port {r['port']} open to internet")
    return errors

all_errors = (validate_cidrs(IPL_INFRA) + validate_tags(IPL_INFRA) +
              validate_s3_security(IPL_INFRA) + validate_sg_no_public_db(IPL_INFRA))
assert all_errors == [], f"Terraform validation failed: {all_errors}"
print("  Terraform validation: CIDR, tags, S3 security, SG — all passed ✓")

# ── Lambda trigger: idempotent ─────────────────────────────────────────────────
PIPELINE_REGISTRY: dict[str, str] = {}  # source_key → run_id

def lambda_trigger(event: dict) -> dict:
    key = event["Records"][0]["s3"]["object"]["key"]
    if not (key.endswith(".csv") and "scorecards" in key):
        return {"action": "skipped", "reason": "not a scorecard"}
    if key in PIPELINE_REGISTRY:
        return {"action": "skipped", "reason": "already_started", "run_id": PIPELINE_REGISTRY[key]}
    run_id = str(uuid.uuid4())[:8]
    PIPELINE_REGISTRY[key] = run_id
    return {"action": "started", "run_id": run_id}

SCORECARD_KEY = "raw/scorecards/2024-04-20/scorecard.csv"
event = {"Records": [{"s3": {"object": {"key": SCORECARD_KEY}}}]}
r1 = lambda_trigger(event)
r2 = lambda_trigger(event)  # duplicate
assert r1["action"] == "started"
assert r2["action"] == "skipped"
assert len(PIPELINE_REGISTRY) == 1
print(f"  Lambda: started={r1['run_id']}, duplicate skipped ✓")
print("Step 1 ✓: Terraform validation and Lambda trigger complete")

Step 2 — Bronze Landing with Iceberg Snapshot and Immutability

Generate 240 raw delivery records for two matches with 2 intentional duplicate records and 10 null bowler rows, write them to the Bronze S3 path with `_ingested_at` and `_source_file` metadata columns, create an Iceberg-style snapshot recording the append operation, and verify that a second write to the same Bronze path raises an immutability error. Assert that Bronze contains all 252 raw rows (240 + 2 duplicates + 10 nulls) — Bronze never filters, it accepts everything from the source.

Analogy🏏Cricket

🏏 Think of it like cricket: Bronze landing is the scorer's box recording the day's play exactly as it happened, including every messy reality, and then sealing the sheet so it can never be quietly altered. Just as the scorer writes down all 252 balls — the two that were accidentally logged twice and the ten where the bowler's name was missed — without silently correcting anything, the code generates 240 records plus 2 duplicates and 10 null-bowler rows and lands all 252, because Bronze never filters; it accepts everything the source gives. Just as each sheet is stamped with the time it was filed and which match it came from, every row carries `_ingested_at` and `_source_file`, and the Iceberg append snapshot logs the operation and exact row count like an entry in the official ledger. Just as a signed, filed scorecard cannot be overwritten, a second write to the same Bronze path raises an immutability error. The payoff: a faithful, provenance-stamped, tamper-proof raw record that every downstream cleaning stage can be audited against.

python

# capstone_bronze_ingestion.py — Step 2: Bronze landing and Iceberg snapshot
import pandas as pd
import numpy as np

# ── Simulated S3 store ────────────────────────────────────────────────────────
S3: dict[str, pd.DataFrame] = {}
ICEBERG_SNAPSHOTS: list[dict] = []

def s3_put(path: str, df: pd.DataFrame) -> None:
    if path in S3:
        raise ValueError(f"Immutability violation: {path} already exists")
    S3[path] = df.copy()

def iceberg_append_snapshot(table: str, path: str, n_rows: int) -> dict:
    snap = {
        "snapshot_id":  len(ICEBERG_SNAPSHOTS) + 1,
        "operation":    "append",
        "table":        table,
        "path":         path,
        "added_records": n_rows,
        "timestamp":    datetime.now(timezone.utc).isoformat(),
    }
    ICEBERG_SNAPSHOTS.append(snap)
    return snap

# ── Generate raw delivery records ─────────────────────────────────────────────
rows = []
for match_id in [10001, 10002]:
    np.random.seed(match_id)
    for i in range(120):
        rows.append({
            "delivery_id":  str(match_id * 1000 + i),
            "match_id":     str(match_id),
            "over":         str((i // 6) + 1),
            "runs_scored":  str(int(np.random.choice([0,1,2,4,6]))),
            "is_wicket":    str(np.random.random() < 0.05),
            "bowler":       np.random.choice(["Bumrah","Shami","Hardik"]),
            "batter":       np.random.choice(["Rohit","Kohli","Gill"]),
        })

# Add 2 duplicates and 10 null bowler rows
duplicates = [{**rows[0], "delivery_id": rows[0]["delivery_id"]},
              {**rows[1], "delivery_id": rows[1]["delivery_id"]}]
null_bowlers = [{**rows[i], "bowler": None} for i in range(2, 12)]

raw_df = pd.DataFrame(rows + duplicates + null_bowlers)
raw_df["_ingested_at"] = datetime.now(timezone.utc).isoformat()
raw_df["_source_file"] = SCORECARD_KEY

BRONZE_PATH = f"s3://ipl-lakehouse-bronze-prod/deliveries/year=2024/month=04/day=20/raw.parquet"
s3_put(BRONZE_PATH, raw_df)
snap1 = iceberg_append_snapshot("bronze.deliveries", BRONZE_PATH, len(raw_df))

assert len(S3[BRONZE_PATH]) == 252  # 240 + 2 dupes + 10 nulls
assert "_ingested_at" in S3[BRONZE_PATH].columns
assert "_source_file" in S3[BRONZE_PATH].columns
assert snap1["operation"] == "append"
assert snap1["added_records"] == 252
print(f"  Bronze: {len(S3[BRONZE_PATH])} rows (raw, no filtering) ✓")
print(f"  Iceberg snapshot {snap1['snapshot_id']}: {snap1['operation']}, {snap1['added_records']} records ✓")

# Immutability: second write to same path must fail
try:
    s3_put(BRONZE_PATH, raw_df)
    assert False, "Should have raised immutability error"
except ValueError:
    print("  Immutability: overwrite blocked ✓")

print("Step 2 ✓: Bronze landing with Iceberg snapshot and immutability complete")

Warning: The Lambda idempotency registry in this exercise uses a Python dictionary that lives only for the duration of the script — in a real Lambda function, the registry would be lost on every cold start and every new Lambda instance. Production idempotency requires a persistent store external to the Lambda function: a DynamoDB table with the source file key as the partition key, or an S3 object whose existence signals that the pipeline has already been triggered for that file. Always implement Lambda idempotency using a persistent external store, not in-memory state.

Extension Challenge: Add a Kinesis-style event stream between the Lambda trigger and the Bronze landing. Publish each raw delivery record as a Kinesis event using the partition key pattern from Module 2 Lesson 11 (match_id as partition key), buffer them in a simulated Event Hubs Capture-style 5-minute window from Module 4 Lesson 23, and write the buffered batch to the Bronze path instead of writing the raw file directly. Assert that the total buffered event count matches the total delivery count, and that all events for the same match_id land in the same partition.