<aside>

Project by: Khaireddine Arbouch

Scalable Data Lakes and Data Warehouses on AWS - Khaireddine Arbouch

</aside>

<aside>

Modern AWS Data Engineering Stack — Breakdown

• AWS S3
• AWS RedShift
• AWS Glue
• AWS EMR
• Amazon Managed Workflows for Apache Airflow (MWAA)
• Apache Hudi

https://youtu.be/rVrTiJRTviA

Component Role Diagram

• Raw Layer Files

End-to-End Pipeline Modules

1. Ingestion + Schema Discovery

Inputs:

• CSVs in: s3://your-bucket/raw/
  • raw_customers.csv, raw_products.csv, raw_transactions.csv

Tool:

• Glue Crawler
  • Crawls S3 → populates Glue Data Catalog
  • Creates tables like raw.customers, raw.products

2. Bronze ETL — Glue Job (Light Cleansing)

Tool:

• AWS Glue + PySpark

Goals:

• Deduplication
• Null value removal
• Format to Parquet

Sample Code (Glue_job.py):

**from pyspark.context import SparkContext
from awsglue.context import GlueContext

glueContext = GlueContext(SparkContext.getOrCreate())
spark = glueContext.spark_session

df = spark.read.csv("s3://your-bucket/raw/raw_customers.csv", header=True)
df_clean = df.dropDuplicates().na.drop()

df_clean.write.parquet("s3://your-bucket/bronze/customers/", mode="overwrite")**

3. Silver ETL — Enrich, Normalize (Glue or EMR)

This can either be in Glue or in PySpark on EMR:

• Add timestamps
• Normalize values
• Join with reference data

**from pyspark.sql.functions import to_timestamp, col

df = spark.read.parquet("s3://your-bucket/bronze/customers/")
df_silver = df.withColumn("signup_date", to_timestamp(col("signup_date")))
df_silver.write.parquet("s3://your-bucket/silver/customers/", mode="overwrite")**

4. Gold ETL — Hudi on EMR (Heavy Lifting)

Tool:

• EMR + Hudi (Apache Hudi library)

Features:

• Upserts & incremental loads
• Partitioning
• Versioning (time-travel queries)

Code (EMR_job.py):

**hudi_options = {
    'hoodie.table.name': 'gold_transactions',
    'hoodie.datasource.write.recordkey.field': 'transaction_id',
    'hoodie.datasource.write.partitionpath.field': 'transaction_date',
    'hoodie.datasource.write.table.name': 'gold_transactions',
    'hoodie.datasource.write.operation': 'upsert',
    'hoodie.datasource.hive.sync.enable': 'true',
}

df.write.format("hudi").options(**hudi_options).mode("overwrite").save("s3://your-bucket/gold/transactions/")**

5. Load to Redshift

Tool:

• Redshift COPY

File: Redshift_job.sql

**COPY reporting.transactions
FROM 's3://your-bucket/gold/transactions/'
IAM_ROLE 'arn:aws:iam::<account>:role/MyRedshiftRole'
FORMAT AS PARQUET;**

6. Orchestration with Airflow (MWAA)

Tool:

• Managed Workflows for Apache Airflow

Code: mwaa_dag_demo.py

**from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def run_glue():
    boto3.client('glue').start_job_run(JobName='bronze_cleanser')

def run_emr():
    # Trigger EMR Step (e.g., using AddJobFlowSteps)

with DAG("end_to_end_etl", start_date=datetime(2023,1,1)) as dag:
    glue_task = PythonOperator(task_id="glue_clean", python_callable=run_glue)
    emr_task = PythonOperator(task_id="emr_transform", python_callable=run_emr)
    glue_task >> emr_task**

Data Lake Zones Summary