Module 1 Hands-On Coding Lab & Practical Scenarios
Test your real-world understanding of LLM fundamentals, token economics, and vector math with these practical Data Engineering scenarios.
🛠️ Challenge 1: The Batch LLM ETL Cost & Token Estimator
Scenario
Your team runs an Airflow DAG that processes 250,000 raw customer feedback comments daily. You are tasked with adding an LLM node that extracts:
sentiment:POSITIVE|NEUTRAL|NEGATIVEissue_category:BILLING|PERFORMANCE|BUG|FEATURE_REQUESTconfidence_score: Float between0.0and1.0
Constraints & Pricing
- Average input text per review: 120 words ($pprox 160$ input tokens).
- System instruction + JSON schema prompt: 80 tokens per request.
- Target JSON output: $pprox 40$ tokens per response.
- Model Pricing (Gemini 1.5 Flash):
- Input: $0.075 per 1,000,000 tokens
- Output: $0.30 per 1,000,000 tokens
Task
Write a Python function calculate_batch_etl_cost(num_records, avg_words_per_record) that outputs:
- Total input tokens
- Total output tokens
- Daily cost in USD ($)
- Estimated monthly cost (30 days)
# Complete the function below:
def calculate_batch_etl_cost(num_records: int, avg_words_per_record: int):
# YOUR CODE HERE
pass
🔍 Click to view Solution & Architecture Explanation
def calculate_batch_etl_cost(num_records: int = 250_000, avg_words_per_record: int = 120):
# Rule of thumb: 1 word ≈ 1.33 tokens (or 1 token ≈ 0.75 words)
tokens_per_review = int(avg_words_per_record * (4 / 3))
system_prompt_tokens = 80
total_input_per_record = tokens_per_review + system_prompt_tokens
total_output_per_record = 40
total_input_tokens = num_records * total_input_per_record
total_output_tokens = num_records * total_output_per_record
# Rates per 1M tokens
input_rate_per_m = 0.075
output_rate_per_m = 0.30
daily_input_cost = (total_input_tokens / 1_000_000) * input_rate_per_m
daily_output_cost = (total_output_tokens / 1_000_000) * output_rate_per_m
total_daily_cost = daily_input_cost + daily_output_cost
monthly_cost = total_daily_cost * 30
return {
"records_processed": f"{num_records:,}",
"total_input_tokens": f"{total_input_tokens:,}",
"total_output_tokens": f"{total_output_tokens:,}",
"daily_cost_usd": f"${total_daily_cost:.2f}",
"monthly_cost_usd": f"${monthly_cost:.2f}"
}
print(calculate_batch_etl_cost())
# Output:
# Total input tokens: 60,000,000
# Total output tokens: 10,000,000
# Daily cost: $4.50 + $3.00 = $7.50
# Monthly cost: $225.00
🛠️ Challenge 2: Semantic Schema Column Matcher
Scenario
You are integrating two legacy warehouse tables from a recent company acquisition:
Source A (Legacy ERP): Columns likecust_nm,tx_amt,dt_created,rcpt_addr.Source B (Standard Medallion DW): Columns likecustomer_full_name,transaction_amount_usd,creation_timestamp,shipping_destination.
Regular string matching (cust_nm vs customer_full_name) fails.
Task
Implement find_best_column_matches(source_cols, target_cols) using simulated or real embeddings with Cosine Similarity:
import numpy as np
def cosine_sim(u, v):
dot = np.dot(u, v)
return dot / (np.linalg.norm(u) * np.linalg.norm(v))
def find_best_column_matches(source_columns, target_columns):
# Match each source column to its highest-affinity target column
pass
🔍 Click to view Solution & Architecture Explanation
import numpy as np
def mock_column_embedder(name: str):
"""
Simulates semantic embedding vectors by hashing word stems.
In real production, pass these names through OpenAI or HuggingFace embeddings.
"""
np.random.seed(abs(hash(name.lower().replace("_", "").replace("cust", "customer").replace("amt", "amount"))) % 2**32)
v = np.random.randn(64)
return v / np.linalg.norm(v)
def find_best_column_matches(source_cols, target_cols):
source_vecs = {c: mock_column_embedder(c) for c in source_cols}
target_vecs = {c: mock_column_embedder(c) for c in target_cols}
matches = []
for s_name, s_vec in source_vecs.items():
best_target = None
best_score = -1.0
for t_name, t_vec in target_vecs.items():
score = float(np.dot(s_vec, t_vec))
if score > best_score:
best_score = score
best_target = t_name
matches.append({
"source_column": s_name,
"matched_target": best_target,
"confidence": round(best_score, 4)
})
return matches
source = ["cust_nm", "tx_amt", "dt_created", "rcpt_addr"]
target = ["customer_full_name", "transaction_amount_usd", "creation_timestamp", "shipping_destination"]
for match in find_best_column_matches(source, target):
print(f"Mapped: {match['source_column']} --> {match['matched_target']} (Score: {match['confidence']})")
🛠️ Challenge 3: Sliding Window Chunking with Token Overlap
Scenario
Before uploading massive technical PDF manuals or database schemas to a vector store (ChromaDB/Pinecone), you must break them into chunks. If you split abruptly by words, crucial sentences that span boundaries get severed, ruining semantic retrieval.
Task
Write a function chunk_text_with_overlap(tokens, chunk_size, chunk_overlap) that takes a list of tokens and yields sliding window chunks:
def chunk_text_with_overlap(tokens: list, chunk_size: int = 100, chunk_overlap: int = 20):
# Output chunks with specified overlap
pass
🔍 Click to view Solution & Architecture Explanation
def chunk_text_with_overlap(tokens: list, chunk_size: int = 100, chunk_overlap: int = 20):
"""
Generate overlapping chunks for robust RAG retrieval.
Stride = chunk_size - chunk_overlap
"""
if chunk_overlap >= chunk_size:
raise ValueError("chunk_overlap must be strictly less than chunk_size")
stride = chunk_size - chunk_overlap
chunks = []
for i in range(0, len(tokens), stride):
chunk = tokens[i : i + chunk_size]
chunks.append(chunk)
if i + chunk_size >= len(tokens):
break
return chunks
# Test verification
test_tokens = list(range(1, 251)) # 250 tokens
chunks = chunk_text_with_overlap(test_tokens, chunk_size=100, chunk_overlap=20)
for idx, c in enumerate(chunks):
print(f"Chunk {idx+1}: Length = {len(c)}, Start = {c[0]}, End = {c[-1]}")
# Chunk 1: [1 ... 100]
# Chunk 2: [81 ... 180] (20 tokens overlapping with Chunk 1!)
# Chunk 3: [161 ... 250] (20 tokens overlapping with Chunk 2!)
Find this content helpful?
☕ Buy me a coffee