{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "b3090aca",
   "metadata": {},
   "source": [
    "# Week 2: ETL, Clean, Normalize and Save (MinIO, Iceberg/Parquet)\n",
    "\n",
    "This notebook demonstrates the PySpark pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d97519f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, mean, stddev\n",
    "\n",
    "print(\"Initializing SparkSession...\")\n",
    "spark = SparkSession.builder \\\n",
    "    .appName(\"Week2_ETL_Iceberg\") \\\n",
    "    .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.12.262,org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.3.1\") \\\n",
    "    .config(\"spark.hadoop.fs.s3a.endpoint\", \"http://minio:9000\") \\\n",
    "    .config(\"spark.hadoop.fs.s3a.access.key\", \"minioadmin\") \\\n",
    "    .config(\"spark.hadoop.fs.s3a.secret.key\", \"minioadmin123\") \\\n",
    "    .config(\"spark.hadoop.fs.s3a.path.style.access\", \"true\") \\\n",
    "    .config(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\") \\\n",
    "    .config(\"spark.sql.extensions\", \"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions\") \\\n",
    "    .config(\"spark.sql.catalog.demo\", \"org.apache.iceberg.spark.SparkCatalog\") \\\n",
    "    .config(\"spark.sql.catalog.demo.type\", \"hadoop\") \\\n",
    "    .config(\"spark.sql.catalog.demo.warehouse\", \"s3a://credit-card/warehouse\") \\\n",
    "    .getOrCreate()\n",
    "    \n",
    "spark.sparkContext.setLogLevel(\"WARN\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "864d04c7",
   "metadata": {},
   "source": [
    "## 1. Read Data from MinIO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a29ad7ea",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(\"s3a://credit-card/creditcard.csv\", header=True, inferSchema=True)\n",
    "print(f\"Original row count: {df.count()}\")\n",
    "df.show(5)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0396abfb",
   "metadata": {},
   "source": [
    "## 2. Clean Data (Drop Duplicates & Nulls)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "da7719a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_clean = df.dropDuplicates().dropna()\n",
    "print(f\"Cleaned row count: {df_clean.count()}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b1863af0",
   "metadata": {},
   "source": [
    "## 3. Normalize Time and Amount Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0cc462d8",
   "metadata": {},
   "outputs": [],
   "source": [
    "stats = df_clean.select(\n",
    "    mean(col('Time')).alias('mean_time'),\n",
    "    stddev(col('Time')).alias('std_time'),\n",
    "    mean(col('Amount')).alias('mean_amount'),\n",
    "    stddev(col('Amount')).alias('std_amount')\n",
    ").collect()[0]\n",
    "\n",
    "df_normalized = df_clean \\\n",
    "    .withColumn('Time', (col('Time') - stats['mean_time']) / stats['std_time']) \\\n",
    "    .withColumn('Amount', (col('Amount') - stats['mean_amount']) / stats['std_amount'])\n",
    "    \n",
    "df_normalized = df_normalized.coalesce(4)\n",
    "df_normalized.select(\"Time\", \"Amount\", \"Class\").show(5)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "625fbd16",
   "metadata": {},
   "source": [
    "## 4. Save to Parquet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "419fd5ea",
   "metadata": {},
   "outputs": [],
   "source": [
    "parquet_path = \"s3a://credit-card/processed/creditcard_clean\"\n",
    "df_normalized.write.mode(\"overwrite\").parquet(parquet_path)\n",
    "print(\"Saved to Parquet\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e7f05ec5",
   "metadata": {},
   "source": [
    "## 5. Save to Apache Iceberg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4c94e943",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"CREATE NAMESPACE IF NOT EXISTS demo.db\")\n",
    "df_normalized.writeTo(\"demo.db.creditcard_clean\").using(\"iceberg\").createOrReplace()\n",
    "print(\"Saved to Iceberg table demo.db.creditcard_clean\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cec8b1f8",
   "metadata": {},
   "source": [
    "## 6. Verify Iceberg Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb0794ee",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_iceberg = spark.table(\"demo.db.creditcard_clean\")\n",
    "print(f\"Iceberg row count: {df_iceberg.count()}\")\n",
    "df_iceberg.show(5)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1a3ba947",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()\n"
   ]
  }
 ],
 "metadata": {},
 "nbformat": 4,
 "nbformat_minor": 5
}