DE_project.py

import requests
import json
import time
import os
from pyspark.sql import SparkSession
from pyspark.sql.functions import (col, upper, trim, regexp_replace, lower, when)
from datetime import datetime

spark = SparkSession.builder.appName("DE_project").getOrCreate()

def fetch_data(base_url, skip):
    url = f"{base_url}{skip}"
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        if data["users"]:
            print(f"Fetching skip={skip}, fetched={len(data['users'])}")
        else:
            data = retry_block(base_url,skip)
    else:
        data = retry_block(base_url,skip)
    return data

def retry_block(base_url,skip):
    retry = 0
    while retry < 3:
        retry += 1
        delay = 2 ** retry
        print(f"Trying again, API request failed for {retry} time")
        time.sleep(delay)
        url = f"{base_url}{skip}"
        response = requests.get(url)
        if response.status_code == 200:
            data = response.json()
            if data["users"]:
                return data
    raise Exception("API failed! Max retry limit reached.")
           
def save_data(data):
    with open("users_raw.json", "w", ) as f:
        json.dump(data, f)

def extract():
    base_url = "https://dummyjson.com/users?limit=40&skip="
    limit = 40
    skip = 0
    total_users = 10
    data_json = []
    print("Extraction started ...")
    while skip<total_users:
        data1 = fetch_data(base_url, skip)
        total_users = data1["total"]
        data_json.extend(data1["users"])
        skip += limit
    rows_fetched = len(data_json)
    print("Rows fetched: ", rows_fetched)
    save_data(data_json)
    print("Extraction completed")

def transform():
    df =  spark.read.json("users_raw.json")
    print("Pipeline Started successfully!")
    raw_count = df.count()
    print(f"rows read: {raw_count}")

    df = df.select(
        col("id").alias("user_id"),
        col("firstName").alias("first_name"),
        col("lastName").alias("last_name"),
        col("age"),
        col("gender"),
        col("address.city").alias("city"),
        col("address.state").alias("state"),
        col("company.name").alias("company_name"),
        col("company.department").alias("department")
    )
    df = df.filter(col("user_id").isNotNull())
    df = df.dropDuplicates(subset=["user_id"])

    df = df.filter(col("first_name").isNotNull())
    df = df.withColumn("first_name", regexp_replace(col("first_name"), "[^a-zA-Z0-9 ]", ""))
    df = df.withColumn("first_name", upper(trim(col("first_name"))))

    df = df.filter(col("last_name").isNotNull())
    df = df.withColumn("last_name", regexp_replace(col("last_name"), "[^a-zA-Z0-9 ]", ""))
    df = df.withColumn("last_name", upper(trim(col("last_name"))))

    df = df.filter(df["age"] > 0)
    df = df.filter(df["age"] < 100)

    df = df.withColumn("gender", lower(trim(col("gender"))))
    df = df.withColumn("gender", when(col("gender").isin("male", "female"), col("gender")).otherwise(None))
    df = df.fillna(value="unknown", subset=["gender", "city", "state", "company_name", "department"])

    df = df.withColumn("city", upper(trim(col("city"))))

    df = df.withColumn("state", upper(trim(col("state"))))

    df = df.withColumn("company_name", trim(col("company_name")))

    df = df.withColumn("department", trim(col("department")))
    final_count = df.count()
    print(f"rows after cleaning: {final_count}")
    return df, raw_count, final_count

def load(df):
    if os.path.isdir("output/users_cleaned"):
        existing_df = spark.read.parquet("output/users_cleaned")
        existing_ids_df = existing_df.select("user_id")
        new_only_df = df.join(existing_ids_df, on="user_id", how="left_anti")
        final_rows = new_only_df.count()
        if final_rows > 0:
            new_only_df.write.mode("append").partitionBy("state").parquet("output/users_cleaned")
            print("File saved successfully!")
            return final_rows
        else:
            print("No new data")
            return final_rows
    else:
        df.write.mode("overwrite").partitionBy("state").parquet("output/users_cleaned")
        print("File saved successfully!")
        count = df.count()
        return count

def audit_log(raw_count, clean_count, new_rows):
    timestamp = datetime.now().isoformat()
    header = "timestamp,raw_rows,cleaned_rows,new_rows_added"
    layout = f"{timestamp},{raw_count},{clean_count},{new_rows}"
    if os.path.exists("output/audit_logs.csv"):
        with open("output/audit_logs.csv", "a") as f:
            f.write(layout + "\n")
    else:
        with open("output/audit_logs.csv", "w") as f:
            f.write(header + "\n")
            f.write(layout + "\n")

def main():
    try:
        extract()
        clean_df, raw_count, clean_count = transform()
        new_rows = load(clean_df)
        audit_log(raw_count, clean_count, new_rows)
        print("Pipeline Completed Successfully!")
    except Exception as e:
        print("Pipeline Failed!" , e)

if __name__ == "__main__":
    main()