AWS Data Engineering – Crash Course

'''
Description: 
Step1: The Glue job will read 3 input data files & create a dataframe for each file- 
		customers
    	products
    	transactions
Step2: Basic string transformations are applied & new load_date column is added
Step3: Write the dataframe into a Hudi table & register in Glue Catalog

This process is equivalent to reading data from input(bronze) layer 
& generate data-lake (silver) data layer

Parameters used in the Glue Job:
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer 
--conf spark.sql.hive.convertMetastoreParquet=false
--datalake-formats hudi
--bucket_name <enter_your_bucket_name>
--ip_path <enter_input_raw_files_path>
--op_path <enter_output_silver_files_path>

Some hardcoded values are used in the code like databasename, tablename.
Feel free to change it as per your requirement.

'''

import sys
from pyspark.sql.functions import *
from pyspark.sql.functions import current_date
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions
from awsglue.transforms import *
from pyspark.context import SparkContext

class ETLSQL:
    def __init__(self):
        params = ["bucket_name","ip_path","op_path"]
        args = getResolvedOptions(sys.argv, params)
        
        self.context = GlueContext(SparkContext.getOrCreate())
        self.spark = self.context.spark_session
        self.job = Job(self.context)
        self.bucket_name = args["bucket_name"]
        self.ip_path = args["ip_path"]
        self.op_path = args["op_path"]
    
    def read_raw_data(self, path: str, format: str = "csv") -> DataFrame:
        """Read raw data from S3"""
        return self.spark.read.format(format) \
            .option("header", "true") \
            .option("inferSchema", "true") \
            .load(path)
    
    def process_customers(self, df: DataFrame) -> DataFrame:
        """Process customer data"""
        return df.withColumn("email", lower(col("email"))) \
            .withColumn("country", upper(col("country"))) \
            .withColumn("load_date", current_date())
    
    def process_products(self, df: DataFrame) -> DataFrame:
        """Process product data"""
        return df.withColumn("category", initcap(col("category"))) \
            .withColumn("subcategory", initcap(col("subcategory"))) \
            .withColumn("load_date", current_date())
    
    def process_transactions(self, df: DataFrame) -> DataFrame:
        """Process transactions data"""
        return df.withColumn("load_date", current_date())
    
    def write_to_hudi(self, df: DataFrame, table_name: str, key_field: str):
        """Write DataFrame to Hudi table"""
        hudi_options = {
            'hoodie.table.name': table_name,
            "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
            'hoodie.datasource.write.recordkey.field': key_field,
            'hoodie.datasource.write.precombine.field': 'load_date',
            'hoodie.datasource.write.operation': 'upsert',
            "hoodie.datasource.write.partitionpath.field": "load_date",
            "hoodie.parquet.compression.codec": "gzip",
            "hoodie.datasource.write.hive_style_partitioning": "true",
            'hoodie.datasource.hive_sync.enable': 'true',
            'hoodie.datasource.hive_sync.database': 'db_etl_sql',
            'hoodie.datasource.hive_sync.table': table_name,
            'hoodie.datasource.hive_sync.partition_fields': 'load_date',
            "hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor",
            "hoodie.datasource.hive_sync.use_jdbc": "false",
            "hoodie.datasource.hive_sync.mode": "hms"
        }
        
        df.write.format('hudi') \
            .options(**hudi_options) \
            .mode('append') \
            .save( f's3://{self.bucket_name}/{self.op_path}/data-lake/{table_name}')

    def run(self):
        # Read raw data
        customers_df = self.read_raw_data(f"s3://{self.bucket_name}/{self.ip_path}/customers/")
        products_df = self.read_raw_data(f"s3://{self.bucket_name}/{self.ip_path}/products/")
        transactions_df = self.read_raw_data(f"s3://{self.bucket_name}/{self.ip_path}/transactions/")

        # Process data
        processed_customers = self.process_customers(customers_df)
        processed_products = self.process_products(products_df)
        processed_transactions = self.process_transactions(transactions_df)
        
        # Write to Hudi tables
        self.write_to_hudi(processed_customers, "customers", "customer_id")
        self.write_to_hudi(processed_products, "products", "product_id")
        self.write_to_hudi(processed_transactions, "transactions", "transaction_id")

        self.job.commit()

if __name__ == "__main__":
    etl = ETLSQL()
    etl.run()

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.functions import count, sum, avg, min, max, col, countDistinct
import sys

""" 
Description:
The EMR job will read data from "data-lake" layer (SILVER) which is generated in the previous step by Glue job & generate "analytics" layer (GOLD).

Step1: Read 3 Hudi tables from "data-lake" path: customers, products & transactions & generate dataframe for each
Step2: Apply transformation to calculate customer_metrics
Step3: Apply transformation to calculate product analytics
Step4: Add new column process_date & write the 2 aggregate metrics into 2 new hudi tables in "analytics" path

Input Parameters:
        a) S3 bucket name
        b) S3 output path for Hudi tables

Script usage: 

spark-submit --deploy-mode cluster 
        --jars /usr/lib/hudi/hudi-spark-bundle.jar --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" 
        --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog" 
        --conf "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension" 
        EMR_job.py <s3_bucket_name> <s3_op_path_for_hudi_table> 

There are some hardcoded values used (like data-lake, analytics path) & you can parameterise & customise it as per your requirement.
"""

class DataLakeProcessor:
    def __init__(self):
        self.spark = SparkSession.builder \
            .appName("DataLakeProcessor") \
            .config('spark.serializer', 'org.apache.spark.serializer.KryoSerializer') \
            .config('spark.sql.hive.convertMetastoreParquet', 'false') \
            .config('hive.metastore.client.factory.class', 'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory') \
            .enableHiveSupport() \
            .getOrCreate()
        self.bucket_name = sys.argv[1]
        self.op_folder = sys.argv[2]
    
    def read_hudi_table(self, table_name: str) -> DataFrame:
        """Read data from Hudi table"""
        return self.spark.read.format('org.apache.hudi') \
            .load(f's3://{self.bucket_name}/{self.op_folder}/data-lake/{table_name}/*')
    
    def calculate_customer_metrics(self) -> DataFrame:
        """Calculate customer purchase metrics"""
        transactions = self.read_hudi_table("transactions")
        
        return transactions.groupBy("customer_id").agg(
                count("transaction_id").alias("total_transactions"),
                sum(col("unit_price") * col("quantity")).alias("total_spent"),
                avg(col("unit_price") * col("quantity")).alias("avg_transaction_amount"),
                min("transaction_date").alias("first_purchase"),
                max("transaction_date").alias("last_purchase")
            )
    
    def create_product_analytics(self) -> DataFrame:
        """Create product analytics"""
        transactions = self.read_hudi_table("transactions")
        products = self.read_hudi_table("products")
        
        return transactions.join(products, "product_id") \
            .groupBy("category", "subcategory") \
            .agg(
                sum("quantity").alias("total_units_sold"),
                sum(col("unit_price") * col("quantity")).alias("total_revenue"),
                avg("unit_price").alias("avg_price"),
                countDistinct("customer_id").alias("unique_customers")
            )
    
    def write_analytics_to_hudi(self, df: DataFrame, table_name: str, key_fields: List[str]):
        """Write analytics results to Hudi"""
        hudi_options = {
            'hoodie.table.name': table_name,
            "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
            'hoodie.datasource.write.recordkey.field': ','.join(key_fields),
            'hoodie.datasource.write.precombine.field': 'process_date',
            "hoodie.datasource.write.partitionpath.field": "process_date",
            "hoodie.parquet.compression.codec": "gzip",
            "hoodie.datasource.write.hive_style_partitioning": "true",
            'hoodie.datasource.write.operation': 'upsert',
            'hoodie.datasource.hive_sync.enable': 'true',
            'hoodie.datasource.hive_sync.database': 'db_etl_sql',
            'hoodie.datasource.hive_sync.table': table_name,
            'hoodie.datasource.hive_sync.partition_fields': 'process_date',
            "hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor",
            "hoodie.datasource.hive_sync.use_jdbc": "false",
            "hoodie.datasource.hive_sync.mode": "hms"
        }

        df.withColumn("process_date", current_date()) \
            .write.format('org.apache.hudi') \
            .options(**hudi_options) \
            .mode('append') \
            .save(f's3://{self.bucket_name}/{self.op_folder}/analytics/{table_name}')

    def run(self):
        # Calculate metrics
        customer_metrics_df = self.calculate_customer_metrics()
        product_analytics_df = self.create_product_analytics()
        
        # Write results
        self.write_analytics_to_hudi(
            customer_metrics_df,
            "customer_metrics",
            ["customer_id"]
        )
        self.write_analytics_to_hudi(
            product_analytics_df,
            "product_analytics",
            ["category", "subcategory"]
        )

if __name__ == "__main__":
    processor = DataLakeProcessor()
    processor.run()

/*
The redshift SQL job will create 5 snapshot tables by reading 2 tables from analytics (GOLD) layer which was generated in the last step by EMR job.
Create an external schema pointing to the glue catalog first.

Kindly replace your IAM Role ARN in the sql command below

Some hardcoded values are used like schema name, databasename, tablename. You can parameterize or customise it as per your requirement.
*/


CREATE EXTERNAL SCHEMA db_etl_sql_ext
FROM DATA CATALOG 
DATABASE 'db_etl_sql' 
IAM_ROLE '<IAM_ROLE_ARN>'
CREATE EXTERNAL DATABASE IF NOT EXISTS;


select * from db_etl_sql_ext.customers;

create schema sch_reporting;

CREATE TABLE sch_reporting.customer_segment_value (
    customer_segment character varying(12) ENCODE zstd distkey,
    customer_count bigint ENCODE zstd,
    avg_spent double precision ENCODE zstd,
    avg_transactions bigint ENCODE zstd,
    spst_date date default CURRENT_DATE
)
DISTSTYLE KEY;

CREATE TABLE sch_reporting.customer_activity_analysis (
    total_customers bigint ENCODE zstd,
    active_90_days bigint ENCODE zstd,
    inactive_90_days bigint ENCODE zstd,
    avg_customer_lifespan bigint ENCODE zstd,
    spst_date date default CURRENT_DATE
)
DISTSTYLE EVEN;

CREATE TABLE sch_reporting.category_performance_analysis  (
    category character varying(100) ENCODE zstd distkey,
    category_revenue double precision ENCODE zstd,
    category_units bigint ENCODE zstd,
    total_customers bigint ENCODE zstd,
    category_avg_price double precision ENCODE zstd,
    revenue_per_customer double precision ENCODE zstd,
    spst_date date default CURRENT_DATE
)
DISTSTYLE KEY;



CREATE TABLE sch_reporting.top_performing_subcategories  (
    category character varying(100) ENCODE zstd distkey,
    subcategory character varying(100) ENCODE zstd,
    total_revenue double precision ENCODE zstd,
    total_units_sold bigint ENCODE zstd,
    category_revenue_share double precision ENCODE zstd,
    spst_date date default CURRENT_DATE
)
DISTSTYLE KEY;




CREATE TABLE sch_reporting.customer_purchase_frequency  (
    frequency_segment character varying(29) ENCODE zstd distkey,
    customer_count bigint ENCODE zstd,
    avg_total_spent double precision ENCODE zstd,
    avg_transaction_value double precision ENCODE zstd,
    spst_date date default CURRENT_DATE
)
DISTSTYLE KEY;



INSERT INTO sch_reporting.customer_segment_value
--customer_segment_value
WITH customer_segments AS (
    SELECT 
        CASE 
            WHEN total_spent >= 1000 THEN 'High Value'
            WHEN total_spent >= 500 THEN 'Medium Value'
            ELSE 'Low Value'
        END as customer_segment,
        COUNT(*) as customer_count,
        AVG(total_spent) as avg_spent,
        AVG(total_transactions) as avg_transactions
    FROM db_etl_sql_ext.customer_metrics
    GROUP BY 
        CASE 
            WHEN total_spent >= 1000 THEN 'High Value'
            WHEN total_spent >= 500 THEN 'Medium Value'
            ELSE 'Low Value'
        END
)
SELECT * FROM customer_segments
ORDER BY avg_spent DESC;


/* this timestamp conversion is required as timestamp is by default stored in number format in Hudi tables */

INSERT INTO sch_reporting.customer_activity_analysis
--customer_activity_analysis
SELECT 
    COUNT(*) as total_customers,
    COUNT(CASE WHEN DATEDIFF('day',TIMESTAMP 'epoch' + last_purchase / 1000000 * INTERVAL '1 second' , CURRENT_DATE) <= 90 THEN 1 END) as active_90_days,
    COUNT(CASE WHEN DATEDIFF('day', TIMESTAMP 'epoch' + last_purchase / 1000000 * INTERVAL '1 second', CURRENT_DATE) > 90 THEN 1 END) as inactive_90_days,
    AVG(DATEDIFF('day', TIMESTAMP 'epoch' + first_purchase / 1000000 * INTERVAL '1 second', TIMESTAMP 'epoch' + last_purchase / 1000000 * INTERVAL '1 second')) as avg_customer_lifespan
FROM db_etl_sql_ext.customer_metrics;


INSERT INTO  sch_reporting.category_performance_analysis
--category_performance_analysis
SELECT 
    category,
    SUM(total_revenue) as category_revenue,
    SUM(total_units_sold) as category_units,
    SUM(unique_customers) as total_customers,
    AVG(avg_price) as category_avg_price,
    SUM(total_revenue)/SUM(unique_customers) as revenue_per_customer
FROM db_etl_sql_ext.product_analytics
GROUP BY category
ORDER BY category_revenue DESC;


INSERT INTO sch_reporting.top_performing_subcategories
--top_performing_subcategories
SELECT 
    category,
    subcategory,
    total_revenue,
    total_units_sold,
    ROUND((total_revenue * 100.0 / SUM(total_revenue) OVER (PARTITION BY category)), 2) as category_revenue_share
FROM db_etl_sql_ext.product_analytics
ORDER BY total_revenue DESC;


INSERT INTO  sch_reporting.customer_purchase_frequency
--customer_purchase_frequency
SELECT 
    CASE 
        WHEN total_transactions >= 10 THEN 'Frequent (10+ transactions)'
        WHEN total_transactions >= 5 THEN 'Regular (5-9 transactions)'
        ELSE 'Occasional (1-4 transactions)'
    END as frequency_segment,
    COUNT(*) as customer_count,
    AVG(total_spent) as avg_total_spent,
    AVG(avg_transaction_amount) as avg_transaction_value
FROM db_etl_sql_ext.customer_metrics
GROUP BY 
    CASE 
        WHEN total_transactions >= 10 THEN 'Frequent (10+ transactions)'
        WHEN total_transactions >= 5 THEN 'Regular (5-9 transactions)'
        ELSE 'Occasional (1-4 transactions)'
    END;

from airflow import DAG
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
from airflow.providers.amazon.aws.operators.emr import EmrAddStepsOperator
from airflow.providers.amazon.aws.sensors.emr import EmrStepSensor
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta
"""
This is airflow DAG which will run Glue job first to read raw data file & generate 3 hudi tables  in data-lake path.
Then it will trigger EMR job which will read 3 hudi tables loaded in previous step to generate 2 hudi tables in analytics path.
Then it will call Redshift Stored Procedure to generate snapshot tables for reporting purpose.

Kindly replace the string values which are enclosed in <> before executing the job.
Moreover create a redshift connection in the Airflow as well
"""
class EtlSqlPipeline:
    def __init__(self):
        self.default_args = {
            'owner': 'etl-sql',
            'depends_on_past': False,
            'start_date': datetime(2025, 1, 1),
            'retries': 1,
            'retry_delay': timedelta(minutes=5)
        }
        self.RS_CONN_ID = '<enter Redshift connection name>'
        self.EMR_cluster_id = '<enter EMR Cluster ID>'

        
        self.dag = DAG(
            dag_id='etl_sql_demo_pipeline',
            default_args=self.default_args,
            schedule_interval=None,
            description="Demo etl-sql pipeline that runs Glue, EMR & Redshift loads",
            catchup=False,
            tags=["etl-sql","demo"],
        )
        
        self.create_tasks()
    
    def create_tasks(self):
        # Start
        start_pipeline = DummyOperator(task_id="LetsGo")

        # Glue ETL Job
        Step1_GlueJob = GlueJobOperator(
            task_id='Step1_GlueJob',
            job_name='<enter glue job name>',
            script_args={
                "--bucket_name": "<enter s3 bucket name>",
                "--ip_path": "<enter input raw data files path>",
                "--op_path": "<enter output hudi table path>",
            },
            dag=self.dag
        )
        
        # EMR Analytics Job
        emr_steps = [{
            'Name': 'Generate_Gold_Layer',
            'ActionOnFailure': 'CONTINUE',
            'HadoopJarStep': {
                'Jar': 'command-runner.jar',
                'Args': [
                    "spark-submit",
                    "--deploy-mode", "cluster", 
                    "--jars", "/usr/lib/hudi/hudi-spark-bundle.jar",
                    "--conf", "spark.serializer=org.apache.spark.serializer.KryoSerializer",
                    "--conf", "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog" ,
                    "--conf", "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension" ,
                    "<enter S3 bucket path for EMR job>",
                    "<enter output S3 bucket name>",
                    "<enter output hudi table S3 path>"
                ]
            }
        }]
        
        Step2_EmrStart = EmrAddStepsOperator(
            task_id='Step2_EmrStart',
            job_flow_id=self.EMR_cluster_id,
            steps=emr_steps,
            dag=self.dag
        )
        
        Step3_EmrWatcher = EmrStepSensor(
            task_id='Step3_EmrWatcher',
            job_flow_id=self.EMR_cluster_id,
            step_id='{{ task_instance.xcom_pull("Step2_EmrStart", key="return_value")[0] }}',
            dag=self.dag
        )
        
        # Redshift Load Tasks
        Step4_RedshiftSpst = PostgresOperator(
            task_id='Step4_RedshiftSpst',
            sql='call sch_reporting.load_snapshot_data()',
            postgres_conn_id=self.RS_CONN_ID,
            dag=self.dag
        )
        
        #End
        end_pipeline = DummyOperator(task_id="YouWon")
        
        # Set dependencies
        start_pipeline >> Step1_GlueJob >> Step2_EmrStart >> Step3_EmrWatcher >> Step4_RedshiftSpst >> end_pipeline

# Create pipeline
pipeline = EtlSqlPipeline()
dag = pipeline.dag