Why This Matters
ETL (Extract, Transform, Load) patterns are fundamental to data engineering. Modern data pipelines require sophisticated patterns for handling slowly changing dimensions, ensuring data quality, and processing incremental data efficiently. Mastering these patterns on AWS services like Glue, EMR, and Lambda is essential for building production-grade data platforms.
Understanding when and how to apply each ETL pattern directly impacts data accuracy, pipeline performance, and operational costs. Poor ETL design leads to data quality issues, missed SLAs, and increased maintenance burden.
Slowly Changing Dimensions (SCDs)
SCDs describe how dimension attributes change over time and how to track those changes for historical analysis.
SCD Type 1: Overwrite
Updates the record in place with no history maintained. Use when you only need current state and history is not required. Example: Correcting a typo in customer name. Simple implementation but cannot analyze historical changes.
SCD Type 2: Versioning
Creates new row for each change, maintaining full history. Use when complete audit trail is required or historical analysis is needed. Example: Tracking customer address changes over time. Full history preserved but table grows over time.
SCD Type 3: Column Versioning
Adds columns to store previous values. Use when you only need to compare current vs. previous. Example: Tracking current and previous sales region. Simple queries with moderate storage but limited to predefined historical values.
SCD Type 6: Hybrid
Combines Type 1 and Type 2 approaches. Overwrites current record while maintaining history. Use when you need both current state and historical tracking.
| SCD Type | Implementation | Use Case | Storage Impact |
|---|---|---|---|
| Type 0 | Fixed | Identity columns | None |
| Type 1 | Overwrite | Current state only | None |
| Type 2 | Row versioning | Full history | High |
| Type 3 | Column version | Limited history | Medium |
| Type 6 | Hybrid | Current + history | Medium-High |
SCD Implementation with AWS Glue
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext
from pyspark.sql import functions as F
from awsglue.dynamicframe import DynamicFrame
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
def implement_scd_type1(source_df, target_df, natural_key):
"""SCD Type 1: Simple overwrite with latest values"""
try:
latest_source = source_df.groupBy(natural_key).agg(
F.max("update_timestamp").alias("max_update_ts")
).join(
source_df,
(source_df[natural_key] == latest_source[natural_key]) &
(source_df.update_timestamp == latest_source.max_update_ts)
)
merged_df = target_df.alias("target").merge(
latest_source.alias("source"),
f"target.{natural_key} = source.{natural_key}"
).whenMatchedUpdateAll(
condition="source.update_timestamp > target.update_timestamp"
).whenNotMatchedInsertAll().execute()
return merged_df
except Exception as e:
print(f"SCD Type 1 error: {str(e)}")
raise
def implement_scd_type2(source_df, target_df, natural_keys, tracked_columns):
"""SCD Type 2: Historical tracking with row versioning"""
try:
source_with_meta = source_df \
.withColumn("row_effective_date", F.current_date()) \
.withColumn("row_expiration_date", F.lit("9999-12-31").cast("date")) \
.withColumn("is_current", F.lit(True)) \
.withColumn("row_version", F.lit(1))
join_condition = " AND ".join([f"target.{k} = source.{k}" for k in natural_keys])
new_records = source_with_meta.alias("source").join(
target_df.alias("target"),
join_condition,
"left_anti"
)
changed_records = source_with_meta.alias("source").join(
target_df.alias("target"),
(join_condition) & (target_df["is_current"] == True),
"inner"
).where(
" OR ".join([f"source.{c} != target.{c}" for c in tracked_columns])
)
expired_records = changed_records.select(
*[f"target.{c}" for c in target_df.columns]
).withColumn("row_expiration_date", F.current_date()) \
.withColumn("is_current", F.lit(False))
new_versions = changed_records.select(
*[f"source.{c}" for c in source_df.columns]
).withColumn("row_effective_date", F.current_date()) \
.withColumn("row_expiration_date", F.lit("9999-12-31").cast("date")) \
.withColumn("is_current", F.lit(True)) \
.withColumn("row_version", target_df["row_version"] + 1)
result = target_df.alias("target").join(
changed_records.select(f"target.{natural_keys[0]}"),
natural_keys[0],
"left_anti"
).union(expired_records) \
.union(new_versions) \
.union(new_records)
return result
except Exception as e:
print(f"SCD Type 2 error: {str(e)}")
raise
# Example usage
source_data = glueContext.create_dynamic_frame.from_catalog(
database="source_db",
table_name="customers"
)
target_data = glueContext.create_dynamic_frame.from_catalog(
database="warehouse",
table_name="dim_customer"
)
source_df = source_data.toDF()
target_df = target_data.toDF()
# SCD Type 1
scd1_result = implement_scd_type1(source_df, target_df, "customer_id")
# SCD Type 2
scd2_result = implement_scd_type2(
source_df, target_df,
["customer_id"],
["address", "phone", "email"]
)
output_df = DynamicFrame.fromDF(scd2_result, glueContext, "output")
glueContext.write_dynamic_frame.from_catalog(
frame=output_df,
database="warehouse",
table_name="dim_customer"
)
Data Quality Framework
Data quality is critical for trustworthy analytics. Implement quality checks at every pipeline stage.
Data Quality Score Formula
Production Data Quality Framework
from awsglue.context import GlueContext
import json
class DataQualityFramework:
"""Comprehensive data quality framework for AWS Glue"""
def __init__(self, glue_context, job_name):
self.glue_context = glue_context
self.job_name = job_name
self.quality_metrics = {}
def validate_schema(self, dynamic_frame, expected_schema):
"""Validate that the data matches expected schema"""
errors = []
for field in expected_schema:
if field['name'] not in dynamic_frame.schema.fieldNames():
errors.append(f"Missing field: {field['name']}")
elif dynamic_frame.schema[field['name']].dataType != field['type']:
errors.append(f"Type mismatch for {field['name']}")
return {
'check': 'schema_validation',
'passed': len(errors) == 0,
'errors': errors,
'record_count': dynamic_frame.count()
}
def check_completeness(self, dynamic_frame, columns):
"""Check for null values in required columns"""
df = dynamic_frame.toDF()
results = []
for col in columns:
null_count = df.filter(df[col].isNull()).count()
total_count = df.count()
completeness = 1 - (null_count / total_count) if total_count > 0 else 0
results.append({
'column': col,
'null_count': null_count,
'completeness_rate': round(completeness * 100, 2),
'passed': completeness >= 0.95
})
return {
'check': 'completeness',
'results': results,
'overall_passed': all(r['passed'] for r in results)
}
def check_accuracy(self, dynamic_frame, rules):
"""Validate business rules for accuracy"""
df = dynamic_frame.toDF()
results = []
for rule in rules:
try:
invalid_count = df.filter(f"NOT ({rule['condition']})").count()
total_count = df.count()
accuracy = 1 - (invalid_count / total_count) if total_count > 0 else 0
results.append({
'rule_name': rule['name'],
'invalid_count': invalid_count,
'accuracy_rate': round(accuracy * 100, 2),
'passed': accuracy >= rule.get('threshold', 0.99)
})
except Exception as e:
results.append({
'rule_name': rule['name'],
'error': str(e),
'passed': False
})
return {
'check': 'accuracy',
'results': results,
'overall_passed': all(r['passed'] for r in results)
}
def check_consistency(self, dynamic_frame, consistency_rules):
"""Check cross-field consistency"""
df = dynamic_frame.toDF()
results = []
for rule in consistency_rules:
try:
violations = df.filter(f"NOT ({rule['condition']})").count()
results.append({
'rule_name': rule['name'],
'violations': violations,
'passed': violations == 0
})
except Exception as e:
results.append({
'rule_name': rule['name'],
'error': str(e),
'passed': False
})
return {
'check': 'consistency',
'results': results,
'overall_passed': all(r['passed'] for r in results)
}
def calculate_quality_score(self, checks):
"""Calculate weighted quality score"""
weights = {
'schema_validation': 0.3,
'completeness': 0.3,
'accuracy': 0.3,
'consistency': 0.1
}
total_score = 0
for check in checks:
check_type = check['check']
weight = weights.get(check_type, 0.1)
if check.get('overall_passed'):
total_score += weight * 100
elif 'results' in check:
passed_count = sum(1 for r in check['results'] if r.get('passed'))
total_score += weight * (passed_count / len(check['results']) * 100)
return round(total_score, 2)
def generate_quality_report(self, checks):
"""Generate comprehensive quality report"""
report = {
'job_name': self.job_name,
'timestamp': str(F.current_timestamp()),
'checks': checks,
'overall_quality_score': self.calculate_quality_score(checks),
'passed': all(c.get('overall_passed', c.get('passed', False)) for c in checks)
}
return report
# Example usage
dq_framework = DataQualityFramework(glueContext, "customer_etl_job")
checks = [
dq_framework.validate_schema(source_data, expected_schema),
dq_framework.check_completeness(source_data, ["customer_id", "name", "email"]),
dq_framework.check_accuracy(source_data, [
{"name": "valid_email", "condition": "email LIKE '%@%.%'", "threshold": 0.99},
{"name": "valid_age", "condition": "age BETWEEN 0 AND 150", "threshold": 0.99}
]),
dq_framework.check_consistency(source_data, [
{"name": "valid_dates", "condition": "start_date <= end_date"}
])
]
report = dq_framework.generate_quality_report(checks)
print(json.dumps(report, indent=2))
Incremental Loading Patterns
Incremental loading processes only new or changed data, dramatically improving efficiency.
Timestamp-Based Loading
Use updated_at column to identify changed records. Track watermark to know last processed timestamp. Process only records newer than watermark. Update watermark after successful processing.
CDC-Based Loading
Use database transaction logs for change capture. Capture inserts, updates, and deletes. AWS DMS handles CDC natively. Provides exactly-once processing semantics.
File-Based Loading
Track processed filenames in checkpoint table. Process only new files. Handle file renaming for idempotency. Use S3 event notifications for triggering.
Production Incremental Loader
from pyspark.sql import functions as F
from delta.tables import DeltaTable
class IncrementalLoader:
"""Handle various incremental loading patterns"""
def __init__(self, glue_context, checkpoint_table="etl_checkpoints"):
self.glue_context = glue_context
self.spark = glue_context.spark_session
self.checkpoint_table = checkpoint_table
def get_watermark(self, job_name, table_name):
"""Get last processed watermark"""
try:
checkpoints = self.spark.read.format("delta").load(
f"s3://data-lake/checkpoints/{self.checkpoint_table}"
)
watermark = checkpoints.filter(
(F.col("job_name") == job_name) &
(F.col("table_name") == table_name)
).select("last_watermark").collect()[0][0]
return watermark
except Exception as e:
print(f"Error getting watermark: {str(e)}")
return "1900-01-01"
def update_watermark(self, job_name, table_name, new_watermark):
"""Update watermark after successful processing"""
try:
checkpoint_df = self.spark.createDataFrame([{
"job_name": job_name,
"table_name": table_name,
"last_watermark": new_watermark,
"updated_at": F.current_timestamp()
}])
checkpoint_df.write.format("delta").mode("append").save(
f"s3://data-lake/checkpoints/{self.checkpoint_table}"
)
except Exception as e:
print(f"Error updating watermark: {str(e)}")
raise
def load_incremental_timestamp(self, source_table, target_table,
timestamp_col="updated_at"):
"""Standard timestamp-based incremental load"""
try:
watermark = self.get_watermark("etl_job", target_table)
incremental_data = self.spark.read.format("delta").load(source_table) \
.filter(F.col(timestamp_col) > watermark)
if incremental_data.count() > 0:
new_watermark = incremental_data.agg(
F.max(timestamp_col)
).collect()[0][0]
self._merge_delta(target_table, incremental_data, "id")
self.update_watermark("etl_job", target_table, new_watermark)
return incremental_data.count()
except Exception as e:
print(f"Error in timestamp incremental load: {str(e)}")
raise
def _merge_delta(self, target_path, source_df, merge_key):
"""Perform Delta Lake merge operation"""
try:
target = DeltaTable.forPath(self.spark, target_path)
target.alias("target").merge(
source_df.alias("source"),
f"target.{merge_key} = source.{merge_key}"
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
except Exception as e:
print(f"Error in Delta merge: {str(e)}")
raise
Common ETL Pitfalls
| Pitfall | Impact | Mitigation |
|---|---|---|
| No idempotency | Duplicate data on retry | Use merge/upsert operations |
| Missing watermarks | Process all data every run | Implement watermark tracking |
| No data quality checks | Bad data in warehouse | Add quality gates at each stage |
| Ignoring schema evolution | Pipeline breaks on changes | Use schema registry |
| Single partition | Shuffle bottleneck | Partition by filter columns |
| No error handling | Silent failures | Implement dead-letter queues |
Performance Considerations
| Factor | Impact | Recommendation |
|---|---|---|
| Partition Strategy | 10-100x query improvement | Partition by date/filter columns |
| File Size | Affects parallelism | Target 128MB-1GB files |
| Join Strategy | 2-10x performance | Broadcast small tables |
| Caching | 5-50x speedup | Cache lookup tables |
| Adaptive Execution | Dynamic optimization | Enable AQE in Spark |
Security Considerations
ETL security requires protecting data at rest and in transit. Encrypt sensitive data before loading. Use IAM roles for service access. Implement column-level security. Audit data access with CloudTrail. Mask PII data in non-production environments. Use Secrets Manager for credentials.
Interview Questions & Answers
Q1: Explain the differences between SCD Type 1, Type 2, and Type 3.
Answer:
SCD Type 1 (Overwrite): Updates the record in place, no history maintained. Use when you only need current state. Example: Correcting a typo in customer name. Simple implementation but cannot analyze historical changes.
SCD Type 2 (Historical): Creates new row for each change, maintains full history. Use when complete audit trail is required. Example: Tracking customer address changes over time. Full history preserved but table grows over time.
SCD Type 3 (Limited History): Adds columns to store previous values. Use when you only need to compare current vs. previous. Example: Tracking current and previous sales region. Simple queries but limited to predefined historical values.
Implementation on AWS:
# SCD Type 1: Simple overwrite
target_df = source_df.dropDuplicates(["natural_key"])
# SCD Type 2: Row versioning
scd2_df = source_df.withColumn("valid_from", F.current_date()) \
.withColumn("valid_to", F.lit("9999-12-31")) \
.withColumn("is_current", F.lit(True))
# SCD Type 3: Column versioning
scd3_df = source_df.withColumn("city_previous", F.lag("city").over(
Window.partitionBy("id").orderBy("updated_at")
))
Q2: How do you handle late-arriving data in an incremental load?
Answer: Implement multiple strategies:
-
Watermark with Buffer Window: Process data with 24-hour buffer. Reprocess recent data periodically.
-
Reprocessing Window: Maintain reprocessing window (e.g., last 7 days). Use Delta Lake time travel for point-in-time recovery.
-
CDC with Transaction Logs: Use database transaction logs instead of timestamps. AWS DMS handles this natively.
-
File-Based Triggers: Process files only when marker files arrive. Use S3 event notifications with Lambda.
-
Idempotent Processing: Design ETL jobs to be safely re-runnable. Use merge/upsert operations instead of append.
Q3: Describe a data quality framework for production ETL.
Answer: Implement multi-layer quality framework:
-
Pre-ingestion Checks: Schema validation, source system health checks, file format validation.
-
In-flight Quality Gates: Completeness checks for nulls, accuracy validation against business rules, consistency checks across fields.
-
Post-load Validation: Row count reconciliation, aggregation comparisons, referential integrity checks.
-
Monitoring and Alerting: CloudWatch custom metrics for quality scores, SNS notifications for threshold breaches.
-
Quality Scoring: Calculate weighted quality score across dimensions. Track trends over time.
Q4: How do you optimize ETL performance on AWS?
Answer: Implement optimization strategies:
-
Partition Strategy: Partition by date and frequently filtered columns. Target 128MB-1GB partition sizes.
-
File Optimization: Use Parquet/ORC with Snappy compression. Control output file sizes with coalesce/repartition.
-
Spark Configuration: Enable adaptive query execution. Configure shuffle partitions based on data volume. Use broadcast joins for small tables.
-
Caching: Cache frequently accessed lookup tables. Use persist for intermediate results.
-
Incremental Processing: Process only changed data with watermarks. Use Delta Lake for upserts.
Q5: Design a real-time ETL pipeline for streaming data.
Answer: Build with Kinesis services:
Architecture:
Kinesis Data Streams -> Kinesis Data Firehose -> S3 (Raw)
|
Kinesis Data Analytics (SQL/Java)
|
S3 (Processed) + Redshift Spectrum
|
QuickSight / Athena
Key Components:
- Kinesis Data Streams for ingestion
- Lambda for real-time transformations
- Kinesis Data Analytics for windowed aggregations
- Firehose for delivery to S3
- Glue for batch processing
Q6: How do you handle schema evolution in ETL pipelines?
Answer: Implement schema management:
-
Delta Lake Schema Evolution: Enable schema auto-merge. Use mergeSchema option for writes.
-
Glue Schema Registry: Use for Avro schemas. Version schemas for compatibility.
-
Best Practices: Add new columns as nullable. Use additive changes only. Version schemas. Test evolution scenarios.
Q7: Explain the trade-offs between Glue and EMR for ETL.
Answer:
| Aspect | AWS Glue | Amazon EMR |
|---|---|---|
| Setup | Serverless, minimal config | Requires cluster management |
| Cost | Pay per DPU-hour | Pay per instance-hour |
| Use Case | ETL jobs, data cataloging | Complex analytics, ML |
| Flexibility | Limited to Spark/Python | Any framework |
| Maintenance | AWS managed | Self-managed or EMR-managed |
Choose Glue for: Simple to moderate ETL, need data catalog, prefer serverless.
Choose EMR for: Complex multi-step processing, need specific frameworks, interactive analysis.
Q8: How do you implement idempotent ETL jobs?
Answer: Ensure jobs can be safely re-run:
-
Merge/Upsert Operations: Use Delta Lake merge for inherently idempotent writes.
-
Checksum-Based Deduplication: Generate checksums for records. Only insert if checksum not exists.
-
Watermark Tracking: Track last successful processing timestamp. Only process data newer than watermark.
-
Transaction Management: Use Delta Lake ACID transactions. Wrap operations in database transactions.
-
File Naming: Use deterministic output file names. Write to temporary location first, then move.