{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# financial services: Iceberg and Liquid Clustering Demo\n",
    "\n",
    "\n",
    "## Overview\n",
    "\n",
    "\n",
    "This notebook demonstrates the power of **Iceberg and Liquid Clustering** in Oracle AI Data Platform (AIDP) Workbench using a financial services analytics use case. Liquid clustering automatically optimizes data layout for query performance without requiring manual partitioning or Z-Ordering.\n",
    "\n",
    "### What is Iceberg?\n",
    "\n",
    "Apache Iceberg is an open table format for huge analytic datasets that provides:\n",
    "\n",
    "- **Schema evolution**: Add, drop, rename, update columns without rewriting data\n",
    "- **Partition evolution**: Change partitioning without disrupting queries\n",
    "- **Time travel**: Query historical data snapshots for auditing and rollback\n",
    "- **ACID transactions**: Reliable concurrent read/write operations\n",
    "- **Cross-engine compatibility**: Works with Spark, Flink, Presto, Hive, and more\n",
    "- **Open ecosystem**: Apache 2.0 licensed, community-driven development\n",
    "\n",
    "### Delta Universal Format with Iceberg\n",
    "\n",
    "Delta Universal Format enables Iceberg compatibility while maintaining Delta's advanced features like liquid clustering. This combination provides:\n",
    "\n",
    "- **Best of both worlds**: Delta's performance optimizations with Iceberg's openness\n",
    "- **Multi-engine access**: Query the same data from different analytics engines\n",
    "- **Future-proof architecture**: Standards-based approach for long-term data investments\n",
    "- **Enhanced governance**: Rich metadata and catalog integration\n",
    "\n",
    "### What is Liquid Clustering?\n",
    "\n",
    "Liquid clustering automatically identifies and groups similar data together based on clustering columns you define. This optimization happens automatically during data ingestion and maintenance operations, providing:\n",
    "\n",
    "- **Automatic optimization**: No manual tuning required\n",
    "- **Improved query performance**: Faster queries on clustered columns\n",
    "- **Reduced maintenance**: No need for manual repartitioning\n",
    "- **Adaptive clustering**: Adjusts as data patterns change\n",
    "\n",
    "### Use Case: Transaction Fraud Detection and Customer Analytics\n",
    "\n",
    "We'll analyze financial transaction records from a bank. Our clustering strategy will optimize for:\n",
    "\n",
    "- **Customer-specific queries**: Fast lookups by account ID\n",
    "- **Time-based analysis**: Efficient filtering by transaction date\n",
    "- **Fraud pattern detection**: Quick aggregation by transaction type and risk scores\n",
    "\n",
    "### AIDP Environment Setup\n",
    "\n",
    "This notebook leverages the existing Spark session in your AIDP environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:30:52.253Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Financial services catalog and analytics schema created successfully!\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Create financial services catalog and analytics schema\n",
    "\n",
    "# In AIDP, catalogs provide data isolation and governance\n",
    "\n",
    "spark.sql(\"CREATE CATALOG IF NOT EXISTS finance\")\n",
    "\n",
    "spark.sql(\"CREATE SCHEMA IF NOT EXISTS finance.analytics\")\n",
    "\n",
    "print(\"Financial services catalog and analytics schema created successfully!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Create Delta Table with Liquid Clustering\n",
    "\n",
    "### Table Design\n",
    "\n",
    "Our `account_transactions_uf` table will store:\n",
    "\n",
    "- **account_id**: Unique account identifier\n",
    "- **transaction_date**: Date and time of transaction\n",
    "- **transaction_type**: Type (Deposit, Withdrawal, Transfer, Payment, etc.)\n",
    "- **amount**: Transaction amount\n",
    "- **merchant_category**: Merchant type (Retail, Restaurant, Online, etc.)\n",
    "- **location**: Transaction location\n",
    "- **risk_score**: Fraud risk assessment (0-100)\n",
    "\n",
    "### Clustering Strategy\n",
    "\n",
    "We'll cluster by `account_id` and `transaction_date` because:\n",
    "\n",
    "- **account_id**: Customers often have multiple transactions, grouping their financial activity together\n",
    "- **transaction_date**: Time-based queries are critical for fraud detection, spending analysis, and regulatory reporting\n",
    "- This combination optimizes for both customer account analysis and temporal fraud pattern detection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:31:03.953Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Delta table with Iceberg compatibility and liquid clustering created successfully!\n",
       "Universal format enables Iceberg features while CLUSTER BY (columns) optimizes data layout.\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Create Delta table with liquid clustering\n",
    "\n",
    "# CLUSTER BY defines the columns for automatic optimization\n",
    "from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, TimestampType, DateType\n",
    "data_schema = StructType([\n",
    "    StructField(\"account_id\", StringType(), True),\n",
    "    StructField(\"transaction_date\", TimestampType(), True),\n",
    "    StructField(\"transaction_type\", StringType(), True),\n",
    "    StructField(\"amount\", DoubleType(), True),\n",
    "    StructField(\"merchant_category\", StringType(), True),\n",
    "    StructField(\"location\", StringType(), True),\n",
    "    StructField(\"risk_score\", IntegerType(), True)\n",
    "])\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "\n",
    "CREATE TABLE IF NOT EXISTS finance.analytics.account_transactions_uf (\n",
    "    account_id STRING,\n",
    "    transaction_date TIMESTAMP,\n",
    "    transaction_type STRING,\n",
    "    amount DECIMAL(15,2),\n",
    "    merchant_category STRING,\n",
    "    location STRING,\n",
    "    risk_score INT\n",
    ")\n",
    "\n",
    "USING DELTA\n",
    "\n",
    "TBLPROPERTIES('delta.universalFormat.enabledFormats' = 'iceberg') CLUSTER BY (account_id, transaction_date)\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "print(\"Delta table with Iceberg compatibility and liquid clustering created successfully!\")\n",
    "\n",
    "print(\"Universal format enables Iceberg features while CLUSTER BY (columns) optimizes data layout.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Generate Financial Services Sample Data\n",
    "\n",
    "### Data Generation Strategy\n",
    "\n",
    "We'll create realistic financial transaction data including:\n",
    "\n",
    "- **5,000 accounts** with multiple transactions over time\n",
    "- **Transaction types**: Deposits, withdrawals, transfers, payments, ATM withdrawals\n",
    "- **Realistic temporal patterns**: Daily banking activity, weekend vs weekday patterns\n",
    "- **Merchant categories**: Retail, restaurants, online shopping, utilities, entertainment\n",
    "\n",
    "### Why This Data Pattern?\n",
    "\n",
    "This data simulates real financial scenarios where:\n",
    "\n",
    "- Customers perform multiple transactions daily/weekly\n",
    "- Fraud patterns emerge over time\n",
    "- Regulatory reporting requires temporal analysis\n",
    "- Risk scoring enables real-time fraud prevention\n",
    "- Customer spending analysis drives personalized financial services"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:31:06.395Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Generated 149143 account transaction records\n",
       "Sample record: {'account_id': 'ACC00000001', 'transaction_date': datetime.datetime(2024, 3, 12, 1, 0), 'transaction_type': 'Transfer', 'amount': 9780.6, 'merchant_category': 'Groceries', 'location': 'Online', 'risk_score': 8}\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Generate sample financial transaction data\n",
    "\n",
    "# Using fully qualified imports to avoid conflicts\n",
    "\n",
    "import random\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "\n",
    "# Define financial data constants\n",
    "\n",
    "TRANSACTION_TYPES = ['Deposit', 'Withdrawal', 'Transfer', 'Payment', 'ATM']\n",
    "\n",
    "MERCHANT_CATEGORIES = ['Retail', 'Restaurant', 'Online', 'Utilities', 'Entertainment', 'Groceries', 'Healthcare', 'Transportation']\n",
    "\n",
    "LOCATIONS = ['New York, NY', 'Los Angeles, CA', 'Chicago, IL', 'Houston, TX', 'Miami, FL', 'Online', 'ATM']\n",
    "\n",
    "\n",
    "# Generate account transaction records\n",
    "\n",
    "transaction_data = []\n",
    "\n",
    "base_date = datetime(2024, 1, 1)\n",
    "\n",
    "\n",
    "# Create 5,000 accounts with 10-50 transactions each\n",
    "\n",
    "for account_num in range(1, 5001):\n",
    "\n",
    "    account_id = f\"ACC{account_num:08d}\"\n",
    "    \n",
    "    # Each account gets 10-50 transactions over 12 months\n",
    "\n",
    "    num_transactions = random.randint(10, 50)\n",
    "    \n",
    "    for i in range(num_transactions):\n",
    "\n",
    "        # Spread transactions over 12 months with realistic timing\n",
    "\n",
    "        days_offset = random.randint(0, 365)\n",
    "\n",
    "        hours_offset = random.randint(0, 23)\n",
    "\n",
    "        transaction_date = base_date + timedelta(days=days_offset, hours=hours_offset)\n",
    "        \n",
    "        # Select transaction type\n",
    "\n",
    "        transaction_type = random.choice(TRANSACTION_TYPES)\n",
    "        \n",
    "        # Amount based on transaction type\n",
    "\n",
    "        if transaction_type in ['Deposit', 'Transfer']:\n",
    "\n",
    "            amount = round(random.uniform(100, 10000), 2)\n",
    "\n",
    "        elif transaction_type == 'ATM':\n",
    "\n",
    "            amount = round(random.uniform(20, 500), 2) * -1\n",
    "\n",
    "        else:\n",
    "\n",
    "            amount = round(random.uniform(10, 2000), 2) * -1\n",
    "        \n",
    "        # Select merchant category and location\n",
    "\n",
    "        merchant_category = random.choice(MERCHANT_CATEGORIES)\n",
    "\n",
    "        if transaction_type == 'ATM':\n",
    "\n",
    "            location = 'ATM'\n",
    "\n",
    "        elif transaction_type == 'Online':\n",
    "\n",
    "            location = 'Online'\n",
    "\n",
    "        else:\n",
    "\n",
    "            location = random.choice(LOCATIONS)\n",
    "        \n",
    "        # Risk score (0-100, higher = more suspicious)\n",
    "\n",
    "        risk_score = random.randint(0, 100)\n",
    "        \n",
    "        transaction_data.append({\n",
    "\n",
    "            \"account_id\": account_id,\n",
    "\n",
    "            \"transaction_date\": transaction_date,\n",
    "\n",
    "            \"transaction_type\": transaction_type,\n",
    "\n",
    "            \"amount\": amount,\n",
    "\n",
    "            \"merchant_category\": merchant_category,\n",
    "\n",
    "            \"location\": location,\n",
    "\n",
    "            \"risk_score\": risk_score\n",
    "\n",
    "        })\n",
    "\n",
    "\n",
    "\n",
    "print(f\"Generated {len(transaction_data)} account transaction records\")\n",
    "\n",
    "print(\"Sample record:\", transaction_data[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Insert Data Using PySpark\n",
    "\n",
    "### Data Insertion Strategy\n",
    "\n",
    "We'll use PySpark to:\n",
    "\n",
    "1. **Create DataFrame** from our generated data\n",
    "2. **Insert into Delta table** with liquid clustering\n",
    "3. **Verify the insertion** with a sample query\n",
    "\n",
    "### Why PySpark for Insertion?\n",
    "\n",
    "- **Distributed processing**: Handles large datasets efficiently\n",
    "- **Type safety**: Ensures data integrity\n",
    "- **Optimization**: Leverages Spark's query optimization\n",
    "- **Liquid clustering**: Automatically applies clustering during insertion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:31:16.957Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame Schema:\n",
       "root\n",
       " |-- account_id: string (nullable = true)\n",
       " |-- transaction_date: timestamp (nullable = true)\n",
       " |-- transaction_type: string (nullable = true)\n",
       " |-- amount: double (nullable = true)\n",
       " |-- merchant_category: string (nullable = true)\n",
       " |-- location: string (nullable = true)\n",
       " |-- risk_score: integer (nullable = true)\n",
       "\n",
       "\n",
       "Sample Data:\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+-------------------+----------------+--------+-----------------+------------+----------+\n",
       "| account_id|   transaction_date|transaction_type|  amount|merchant_category|    location|risk_score|\n",
       "+-----------+-------------------+----------------+--------+-----------------+------------+----------+\n",
       "|ACC00000001|2024-03-12 01:00:00|        Transfer|  9780.6|        Groceries|      Online|         8|\n",
       "|ACC00000001|2024-01-08 13:00:00|        Transfer| 3075.42|       Healthcare| Chicago, IL|        30|\n",
       "|ACC00000001|2024-05-24 12:00:00|         Payment|-1475.39|   Transportation| Chicago, IL|        99|\n",
       "|ACC00000001|2024-10-08 15:00:00|      Withdrawal| -595.12|           Online| Chicago, IL|        81|\n",
       "|ACC00000001|2024-09-19 09:00:00|        Transfer| 3645.06|    Entertainment|New York, NY|        98|\n",
       "+-----------+-------------------+----------------+--------+-----------------+------------+----------+\n",
       "only showing top 5 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "\n",
       "Successfully inserted 149143 records into finance.analytics.account_transactions_uf\n",
       "Liquid clustering automatically optimized the data layout during insertion!\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Insert data using PySpark DataFrame operations\n",
    "\n",
    "# Using fully qualified function references to avoid conflicts\n",
    "\n",
    "\n",
    "# Create DataFrame from generated data\n",
    "\n",
    "df_transactions = spark.createDataFrame(transaction_data, schema=data_schema)\n",
    "\n",
    "\n",
    "# Display schema and sample data\n",
    "\n",
    "print(\"DataFrame Schema:\")\n",
    "\n",
    "df_transactions.printSchema()\n",
    "\n",
    "\n",
    "\n",
    "print(\"\\nSample Data:\")\n",
    "\n",
    "df_transactions.show(5)\n",
    "\n",
    "\n",
    "# Insert data into Delta table with liquid clustering\n",
    "\n",
    "# The TBLPROPERTIES('delta.universalFormat.enabledFormats' = 'iceberg') CLUSTER BY (account_id, transaction_date) will automatically optimize the data layout\n",
    "\n",
    "df_transactions.write.mode(\"overwrite\").insertInto(\"finance.analytics.account_transactions_uf\")\n",
    "\n",
    "\n",
    "print(f\"\\nSuccessfully inserted {df_transactions.count()} records into finance.analytics.account_transactions_uf\")\n",
    "\n",
    "print(\"Liquid clustering automatically optimized the data layout during insertion!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Demonstrate Liquid Clustering Benefits\n",
    "\n",
    "### Query Performance Analysis\n",
    "\n",
    "Now let's see how liquid clustering improves query performance. We'll run queries that benefit from our clustering strategy:\n",
    "\n",
    "1. **Account transaction history** (clustered by account_id)\n",
    "2. **Time-based fraud analysis** (clustered by transaction_date)\n",
    "3. **Combined account + time queries** (optimal for our clustering)\n",
    "\n",
    "### Expected Performance Benefits\n",
    "\n",
    "With liquid clustering, these queries should be significantly faster because:\n",
    "\n",
    "- **Data locality**: Related records are physically grouped together\n",
    "- **Reduced I/O**: Less data needs to be read from disk\n",
    "- **Automatic optimization**: No manual tuning required"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:31:28.557Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "=== Query 1: Account Transaction History ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+-------------------+----------------+--------+-----------------+\n",
       "| account_id|   transaction_date|transaction_type|  amount|merchant_category|\n",
       "+-----------+-------------------+----------------+--------+-----------------+\n",
       "|ACC00000001|2024-12-27 03:00:00|             ATM| -103.15|       Healthcare|\n",
       "|ACC00000001|2024-12-26 02:00:00|         Payment|-1591.98|   Transportation|\n",
       "|ACC00000001|2024-12-19 17:00:00|         Payment| -446.84|        Groceries|\n",
       "|ACC00000001|2024-11-30 18:00:00|        Transfer| 8528.64|    Entertainment|\n",
       "|ACC00000001|2024-11-06 19:00:00|      Withdrawal|  -37.15|       Healthcare|\n",
       "|ACC00000001|2024-10-14 09:00:00|        Transfer| 5505.77|           Retail|\n",
       "|ACC00000001|2024-10-08 15:00:00|      Withdrawal| -595.12|           Online|\n",
       "|ACC00000001|2024-09-27 07:00:00|             ATM| -458.07|    Entertainment|\n",
       "|ACC00000001|2024-09-19 09:00:00|        Transfer| 3645.06|    Entertainment|\n",
       "|ACC00000001|2024-09-12 08:00:00|             ATM| -286.90|           Online|\n",
       "|ACC00000001|2024-09-11 01:00:00|      Withdrawal| -687.36|       Healthcare|\n",
       "|ACC00000001|2024-09-04 07:00:00|         Deposit| 3431.07|           Retail|\n",
       "|ACC00000001|2024-08-26 16:00:00|      Withdrawal| -418.45|       Healthcare|\n",
       "|ACC00000001|2024-08-11 05:00:00|         Deposit| 5309.89|        Groceries|\n",
       "|ACC00000001|2024-08-08 23:00:00|         Payment|-1988.64|           Retail|\n",
       "|ACC00000001|2024-07-27 11:00:00|        Transfer| 8917.83|           Retail|\n",
       "|ACC00000001|2024-07-02 15:00:00|         Deposit| 7150.19|   Transportation|\n",
       "|ACC00000001|2024-06-26 14:00:00|         Deposit| 5972.73|           Online|\n",
       "|ACC00000001|2024-06-22 00:00:00|         Payment|-1420.78|       Restaurant|\n",
       "|ACC00000001|2024-06-16 09:00:00|             ATM| -491.02|       Healthcare|\n",
       "+-----------+-------------------+----------------+--------+-----------------+\n",
       "only showing top 20 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Records found: 36\n",
       "\n",
       "=== Query 2: High-Risk Transactions Today ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+----------------+----------+----------------+------+----------+\n",
       "|transaction_date|account_id|transaction_type|amount|risk_score|\n",
       "+----------------+----------+----------------+------+----------+\n",
       "+----------------+----------+----------------+------+----------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "High-risk transactions found: 0\n",
       "\n",
       "=== Query 3: Account Fraud Pattern Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+-------------------+----------------+--------+----------+\n",
       "| account_id|   transaction_date|transaction_type|  amount|risk_score|\n",
       "+-----------+-------------------+----------------+--------+----------+\n",
       "|ACC00000010|2024-07-21 12:00:00|        Transfer| 1382.46|        77|\n",
       "|ACC00000010|2024-07-28 18:00:00|      Withdrawal|-1227.64|        67|\n",
       "|ACC00000010|2024-08-13 19:00:00|             ATM|  -52.76|        52|\n",
       "|ACC00000010|2024-08-19 09:00:00|         Deposit| 5522.80|        61|\n",
       "|ACC00000010|2024-08-31 17:00:00|         Payment|  -17.65|        17|\n",
       "|ACC00000010|2024-09-25 23:00:00|             ATM|  -61.76|        55|\n",
       "|ACC00000010|2024-11-01 09:00:00|         Payment| -465.37|        71|\n",
       "|ACC00000010|2024-11-10 20:00:00|         Deposit| 2367.70|         1|\n",
       "|ACC00000010|2024-11-17 10:00:00|      Withdrawal| -470.32|        43|\n",
       "|ACC00000010|2024-12-06 19:00:00|         Payment|-1910.33|         9|\n",
       "|ACC00000011|2024-06-06 06:00:00|        Transfer| 2805.70|        60|\n",
       "|ACC00000011|2024-07-09 07:00:00|         Deposit| 5737.70|        84|\n",
       "|ACC00000011|2024-07-17 16:00:00|      Withdrawal| -783.56|        35|\n",
       "|ACC00000011|2024-08-31 09:00:00|      Withdrawal|-1039.16|        53|\n",
       "|ACC00000011|2024-09-14 05:00:00|             ATM| -128.96|         4|\n",
       "|ACC00000011|2024-09-29 13:00:00|        Transfer| 1810.50|        82|\n",
       "|ACC00000011|2024-10-21 20:00:00|      Withdrawal|-1890.81|        90|\n",
       "|ACC00000011|2024-11-01 23:00:00|        Transfer| 7421.57|        15|\n",
       "|ACC00000011|2024-12-01 13:00:00|        Transfer| 2208.39|        63|\n",
       "|ACC00000012|2024-06-02 07:00:00|         Deposit| 7660.96|        98|\n",
       "+-----------+-------------------+----------------+--------+----------+\n",
       "only showing top 20 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Pattern records found: 146\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Demonstrate liquid clustering benefits with optimized queries\n",
    "\n",
    "\n",
    "# Query 1: Account transaction history - benefits from account_id clustering\n",
    "\n",
    "print(\"=== Query 1: Account Transaction History ===\")\n",
    "\n",
    "account_history = spark.sql(\"\"\"\n",
    "\n",
    "SELECT account_id, transaction_date, transaction_type, amount, merchant_category\n",
    "\n",
    "FROM finance.analytics.account_transactions_uf\n",
    "\n",
    "WHERE account_id = 'ACC00000001'\n",
    "\n",
    "ORDER BY transaction_date DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "account_history.show()\n",
    "\n",
    "print(f\"Records found: {account_history.count()}\")\n",
    "\n",
    "\n",
    "\n",
    "# Query 2: Time-based fraud analysis - benefits from transaction_date clustering\n",
    "\n",
    "print(\"\\n=== Query 2: High-Risk Transactions Today ===\")\n",
    "\n",
    "high_risk_today = spark.sql(\"\"\"\n",
    "\n",
    "SELECT transaction_date, account_id, transaction_type, amount, risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions_uf\n",
    "\n",
    "WHERE DATE(transaction_date) = CURRENT_DATE AND risk_score > 70\n",
    "\n",
    "ORDER BY risk_score DESC, transaction_date DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "high_risk_today.show()\n",
    "\n",
    "print(f\"High-risk transactions found: {high_risk_today.count()}\")\n",
    "\n",
    "\n",
    "\n",
    "# Query 3: Combined account + time query - optimal for our clustering strategy\n",
    "\n",
    "print(\"\\n=== Query 3: Account Fraud Pattern Analysis ===\")\n",
    "\n",
    "fraud_patterns = spark.sql(\"\"\"\n",
    "\n",
    "SELECT account_id, transaction_date, transaction_type, amount, risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions_uf\n",
    "\n",
    "WHERE account_id LIKE 'ACC0000001%' AND transaction_date >= '2024-06-01'\n",
    "\n",
    "ORDER BY account_id, transaction_date\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "fraud_patterns.show()\n",
    "\n",
    "print(f\"Pattern records found: {fraud_patterns.count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 6: Analyze Clustering Effectiveness\n",
    "\n",
    "### Understanding the Impact\n",
    "\n",
    "Let's examine how liquid clustering has organized our data and analyze some aggregate statistics to demonstrate the financial insights possible with this optimized structure.\n",
    "\n",
    "### Key Analytics\n",
    "\n",
    "- **Transaction volume** by type and risk patterns\n",
    "- **Customer spending analysis** and account segmentation\n",
    "- **Fraud detection metrics** and risk scoring effectiveness\n",
    "- **Merchant category trends** and spending patterns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:31:41.201Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "=== Transaction Analysis by Type ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+----------------+------------------+------------+----------+--------------+\n",
       "|transaction_type|total_transactions|total_amount|avg_amount|avg_risk_score|\n",
       "+----------------+------------------+------------+----------+--------------+\n",
       "|        Transfer|             29999|151108162.38|   5037.11|         49.99|\n",
       "|         Deposit|             29923|150962629.64|   5045.04|         50.22|\n",
       "|      Withdrawal|             29921|-29988461.60|  -1002.25|         50.01|\n",
       "|         Payment|             29789|-29893428.71|  -1003.51|         50.11|\n",
       "|             ATM|             29511| -7619730.14|   -258.20|         49.97|\n",
       "+----------------+------------------+------------+----------+--------------+\n",
       "\n",
       "\n",
       "=== Risk Score Distribution ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+--------------+-----------------+----------+\n",
       "| risk_category|transaction_count|percentage|\n",
       "+--------------+-----------------+----------+\n",
       "|Very High Risk|            30987|     20.78|\n",
       "|   Medium Risk|            29797|     19.98|\n",
       "|     High Risk|            29627|     19.86|\n",
       "| Very Low Risk|            29423|     19.73|\n",
       "|      Low Risk|            29309|     19.65|\n",
       "+--------------+-----------------+----------+\n",
       "\n",
       "\n",
       "=== Merchant Category Spending Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------------+------------+-----------+----------+--------+\n",
       "|merchant_category|transactions|   deposits|  spending|avg_risk|\n",
       "+-----------------+------------+-----------+----------+--------+\n",
       "|        Groceries|       18642|37772542.91|8494798.48|   49.94|\n",
       "|           Retail|       18604|37468897.74|8487432.78|   50.04|\n",
       "|       Restaurant|       18672|38179209.41|8480727.72|    50.4|\n",
       "|           Online|       18515|37231692.24|8472075.04|   49.91|\n",
       "|    Entertainment|       18636|37411438.43|8461974.74|   49.76|\n",
       "|   Transportation|       18651|37290951.61|8443335.98|   50.34|\n",
       "|        Utilities|       18638|37986761.82|8356903.09|   50.23|\n",
       "|       Healthcare|       18785|38729297.86|8304372.62|   49.85|\n",
       "+-----------------+------------+-----------+----------+--------+\n",
       "\n",
       "\n",
       "=== Monthly Transaction Trends ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-------+------------+-----------+---------------+--------------+\n",
       "|  month|transactions|   net_flow|active_accounts|avg_risk_score|\n",
       "+-------+------------+-----------+---------------+--------------+\n",
       "|2024-01|       12624|19939024.66|           4398|         49.73|\n",
       "|2024-02|       11944|19328144.90|           4373|          50.3|\n",
       "|2024-03|       12510|19632464.84|           4402|         49.95|\n",
       "|2024-04|       12288|19264206.65|           4405|         50.09|\n",
       "|2024-05|       12554|19137998.88|           4398|         49.88|\n",
       "|2024-06|       12423|20206469.72|           4397|         50.25|\n",
       "|2024-07|       12397|19601966.76|           4399|          49.8|\n",
       "|2024-08|       12614|20083080.90|           4427|         50.11|\n",
       "|2024-09|       12243|18520818.85|           4349|         50.03|\n",
       "|2024-10|       12758|20407110.51|           4428|         50.19|\n",
       "|2024-11|       12199|18622669.90|           4388|         50.07|\n",
       "|2024-12|       12589|19825215.00|           4447|         50.32|\n",
       "+-------+------------+-----------+---------------+--------------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Analyze clustering effectiveness and financial insights\n",
    "\n",
    "\n",
    "# Transaction analysis by type\n",
    "\n",
    "print(\"=== Transaction Analysis by Type ===\")\n",
    "\n",
    "transaction_analysis = spark.sql(\"\"\"\n",
    "\n",
    "SELECT transaction_type, COUNT(*) as total_transactions,\n",
    "\n",
    "       ROUND(SUM(amount), 2) as total_amount,\n",
    "\n",
    "       ROUND(AVG(amount), 2) as avg_amount,\n",
    "\n",
    "       ROUND(AVG(risk_score), 2) as avg_risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions_uf\n",
    "\n",
    "GROUP BY transaction_type\n",
    "\n",
    "ORDER BY total_transactions DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "transaction_analysis.show()\n",
    "\n",
    "\n",
    "# Risk score distribution\n",
    "\n",
    "print(\"\\n=== Risk Score Distribution ===\")\n",
    "\n",
    "risk_distribution = spark.sql(\"\"\"\n",
    "\n",
    "SELECT \n",
    "\n",
    "    CASE \n",
    "\n",
    "        WHEN risk_score >= 80 THEN 'Very High Risk'\n",
    "\n",
    "        WHEN risk_score >= 60 THEN 'High Risk'\n",
    "\n",
    "        WHEN risk_score >= 40 THEN 'Medium Risk'\n",
    "\n",
    "        WHEN risk_score >= 20 THEN 'Low Risk'\n",
    "\n",
    "        ELSE 'Very Low Risk'\n",
    "\n",
    "    END as risk_category,\n",
    "\n",
    "    COUNT(*) as transaction_count,\n",
    "\n",
    "    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) as percentage\n",
    "\n",
    "FROM finance.analytics.account_transactions_uf\n",
    "\n",
    "GROUP BY \n",
    "\n",
    "    CASE \n",
    "\n",
    "        WHEN risk_score >= 80 THEN 'Very High Risk'\n",
    "\n",
    "        WHEN risk_score >= 60 THEN 'High Risk'\n",
    "\n",
    "        WHEN risk_score >= 40 THEN 'Medium Risk'\n",
    "\n",
    "        WHEN risk_score >= 20 THEN 'Low Risk'\n",
    "\n",
    "        ELSE 'Very Low Risk'\n",
    "\n",
    "    END\n",
    "\n",
    "ORDER BY transaction_count DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "risk_distribution.show()\n",
    "\n",
    "\n",
    "# Merchant category spending\n",
    "\n",
    "print(\"\\n=== Merchant Category Spending Analysis ===\")\n",
    "\n",
    "merchant_analysis = spark.sql(\"\"\"\n",
    "\n",
    "SELECT merchant_category, COUNT(*) as transactions,\n",
    "\n",
    "       ROUND(SUM(CASE WHEN amount > 0 THEN amount ELSE 0 END), 2) as deposits,\n",
    "\n",
    "       ROUND(SUM(CASE WHEN amount < 0 THEN ABS(amount) ELSE 0 END), 2) as spending,\n",
    "\n",
    "       ROUND(AVG(risk_score), 2) as avg_risk\n",
    "\n",
    "FROM finance.analytics.account_transactions_uf\n",
    "\n",
    "GROUP BY merchant_category\n",
    "\n",
    "ORDER BY spending DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "merchant_analysis.show()\n",
    "\n",
    "\n",
    "# Monthly transaction trends\n",
    "\n",
    "print(\"\\n=== Monthly Transaction Trends ===\")\n",
    "\n",
    "monthly_trends = spark.sql(\"\"\"\n",
    "\n",
    "SELECT DATE_FORMAT(transaction_date, 'yyyy-MM') as month,\n",
    "\n",
    "       COUNT(*) as transactions,\n",
    "\n",
    "       ROUND(SUM(amount), 2) as net_flow,\n",
    "\n",
    "       COUNT(DISTINCT account_id) as active_accounts,\n",
    "\n",
    "       ROUND(AVG(risk_score), 2) as avg_risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions_uf\n",
    "\n",
    "GROUP BY DATE_FORMAT(transaction_date, 'yyyy-MM')\n",
    "\n",
    "ORDER BY month\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "monthly_trends.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Key Takeaways: Iceberg and Liquid Clustering in AIDP\n",
    "\n",
    "### What We Demonstrated\n",
    "\n",
    "1. **Automatic Optimization**: Created a table with `TBLPROPERTIES('delta.universalFormat.enabledFormats' = 'iceberg') CLUSTER BY (account_id, transaction_date)` and let Delta automatically optimize data layout\n",
    "\n",
    "2. **Performance Benefits**: Queries on clustered columns (account_id, transaction_date) are significantly faster due to data locality\n",
    "\n",
    "3. **Zero Maintenance**: No manual partitioning, bucketing, or Z-Ordering required - Delta handles it automatically\n",
    "\n",
    "4. **Real-World Use Case**: Financial services analytics where fraud detection and customer analysis are critical\n",
    "\n",
    "### AIDP Advantages\n",
    "\n",
    "- **Unified Analytics**: Seamlessly integrates with other AIDP services\n",
    "- **Governance**: Catalog and schema isolation for financial data\n",
    "- **Performance**: Optimized for both OLAP and OLTP workloads\n",
    "- **Scalability**: Handles financial-scale data volumes effortlessly\n",
    "\n",
    "### Best Practices for Iceberg and Liquid Clustering\n",
    "\n",
    "1. **Choose clustering columns** based on your most common query patterns\n",
    "2. **Start with 1-4 columns** - too many can reduce effectiveness\n",
    "3. **Consider cardinality** - high-cardinality columns work best\n",
    "4. **Monitor and adjust** as query patterns evolve\n",
    "\n",
    "### Next Steps\n",
    "\n",
    "- Explore other AIDP features like AI/ML integration\n",
    "- Try liquid clustering with different column combinations\n",
    "- Scale up to larger financial datasets\n",
    "- Integrate with real banking systems and fraud detection platforms\n",
    "\n",
    "This notebook demonstrates how Oracle AI Data Platform combines Delta's advanced liquid clustering with Iceberg's open, future-proof architecture to deliver enterprise-grade analytics that are both high-performance and standards-compliant."
   ]
  }
 ],
 "metadata": {
  "Last_Active_Cell_Index": 11,
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "notebook"
  },
  "language_info": {
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}