{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Healthcare: Iceberg and Liquid Clustering Demo\n",
    "\n",
    "## Overview\n",
    "\n",
    "This notebook demonstrates the power of **Iceberg and Liquid Clustering** in Oracle AI Data Platform (AIDP) Workbench using a healthcare analytics use case. Liquid clustering automatically optimizes data layout for query performance without requiring manual partitioning or Z-Ordering, now enhanced with Iceberg compatibility through Delta Universal Format.\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: Patient Diagnosis Analytics\n",
    "\n",
    "We'll analyze patient diagnosis records from a healthcare system. Our clustering strategy will optimize for:\n",
    "- **Patient-specific queries**: Fast lookups by patient ID\n",
    "- **Time-based analysis**: Efficient filtering by diagnosis date\n",
    "- **Diagnosis patterns**: Quick aggregation by diagnosis type\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:35:47.805Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Healthcare catalog and gold schema created successfully!\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Create healthcare catalog and gold schema\n",
    "# In AIDP, catalogs provide data isolation and governance\n",
    "\n",
    "spark.sql(\"CREATE CATALOG IF NOT EXISTS healthcare\")\n",
    "spark.sql(\"CREATE SCHEMA IF NOT EXISTS healthcare.gold\")\n",
    "\n",
    "print(\"Healthcare catalog and gold schema created successfully!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Create Delta Table with Liquid Clustering\n",
    "\n",
    "### Table Design\n",
    "\n",
    "Our `patient_diagnoses_uf` table will store:\n",
    "- **patient_id**: Unique patient identifier\n",
    "- **diagnosis_date**: When the diagnosis was made\n",
    "- **diagnosis_code**: ICD-10 diagnosis code\n",
    "- **diagnosis_description**: Human-readable diagnosis\n",
    "- **severity_level**: Critical, High, Medium, Low\n",
    "- **treating_physician**: Physician ID\n",
    "- **facility_id**: Healthcare facility\n",
    "\n",
    "### Clustering Strategy\n",
    "\n",
    "We'll cluster by `patient_id` and `diagnosis_date` because:\n",
    "- **patient_id**: Patients often have multiple visits, grouping their records together\n",
    "- **diagnosis_date**: Time-based queries are common in healthcare analytics\n",
    "- This combination optimizes for both patient history lookups and temporal analysis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:36:00.504Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Delta table with Iceberg compatibility and liquid clustering created successfully!\n",
       "Universal format enables Iceberg features while CLUSTER BY (patient_id, diagnosis_date) optimizes data layout.\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Create Delta table with Iceberg compatibility via Universal Format and liquid clustering\n",
    "# TBLPROPERTIES enables Delta Universal Format for Iceberg compatibility\n",
    "\n",
    "from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, TimestampType, DateType\n",
    "data_schema = StructType([\n",
    "    StructField(\"patient_id\", StringType(), True),\n",
    "    StructField(\"diagnosis_date\", DateType(), True),\n",
    "    StructField(\"diagnosis_code\", StringType(), True),\n",
    "    StructField(\"diagnosis_description\", StringType(), True),\n",
    "    StructField(\"severity_level\", StringType(), True),\n",
    "    StructField(\"treating_physician\", StringType(), True),\n",
    "    StructField(\"facility_id\", StringType(), True)\n",
    "])\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "CREATE TABLE IF NOT EXISTS healthcare.gold.patient_diagnoses_uf (\n",
    "    patient_id STRING,\n",
    "    diagnosis_date DATE,\n",
    "    diagnosis_code STRING,\n",
    "    diagnosis_description STRING,\n",
    "    severity_level STRING,\n",
    "    treating_physician STRING,\n",
    "    facility_id STRING\n",
    ")\n",
    "USING DELTA\n",
    "TBLPROPERTIES('delta.universalFormat.enabledFormats' = 'iceberg')\n",
    "CLUSTER BY (patient_id, diagnosis_date)\n",
    "\"\"\")\n",
    "\n",
    "print(\"Delta table with Iceberg compatibility and liquid clustering created successfully!\")\n",
    "print(\"Universal format enables Iceberg features while CLUSTER BY (patient_id, diagnosis_date) optimizes data layout.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Generate Healthcare Sample Data\n",
    "\n",
    "### Data Generation Strategy\n",
    "\n",
    "We'll create realistic healthcare data including:\n",
    "- **100 patients** with multiple diagnoses over time\n",
    "- **Common diagnoses**: Diabetes, Hypertension, Asthma, etc.\n",
    "- **Realistic temporal patterns**: Follow-up visits, chronic condition management\n",
    "- **Multiple facilities**: Different hospitals/clinics\n",
    "\n",
    "### Why This Data Pattern?\n",
    "\n",
    "This data simulates real healthcare scenarios where:\n",
    "- Patients have multiple encounters\n",
    "- Chronic conditions require ongoing monitoring\n",
    "- Time-based analysis reveals treatment effectiveness\n",
    "- Facility-level reporting is needed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:36:02.927Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Generated 349 patient diagnosis records\n",
       "Sample record: {'patient_id': 'PAT0001', 'diagnosis_date': datetime.date(2024, 4, 25), 'diagnosis_code': 'Z51.11', 'diagnosis_description': 'Encounter for antineoplastic chemotherapy', 'severity_level': 'Critical', 'treating_physician': 'DR_WILLIAMS', 'facility_id': 'HOSP002'}\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Generate sample healthcare diagnosis data\n",
    "# Using fully qualified pyspark.sql.functions to avoid conflicts\n",
    "\n",
    "import random\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "# Define healthcare data constants\n",
    "DIAGNOSES = [\n",
    "    (\"E11.9\", \"Type 2 diabetes mellitus without complications\", \"Medium\"),\n",
    "    (\"I10\", \"Essential hypertension\", \"High\"),\n",
    "    (\"J45.909\", \"Unspecified asthma, uncomplicated\", \"Medium\"),\n",
    "    (\"M54.5\", \"Low back pain\", \"Low\"),\n",
    "    (\"N39.0\", \"Urinary tract infection, site not specified\", \"Medium\"),\n",
    "    (\"Z51.11\", \"Encounter for antineoplastic chemotherapy\", \"Critical\"),\n",
    "    (\"I25.10\", \"Atherosclerotic heart disease of native coronary artery without angina pectoris\", \"High\"),\n",
    "    (\"F41.9\", \"Anxiety disorder, unspecified\", \"Medium\"),\n",
    "    (\"M79.3\", \"Panniculitis, unspecified\", \"Low\"),\n",
    "    (\"Z00.00\", \"Encounter for general adult medical examination without abnormal findings\", \"Low\")\n",
    "]\n",
    "\n",
    "FACILITIES = [\"HOSP001\", \"HOSP002\", \"CLINIC001\", \"CLINIC002\", \"URGENT001\"]\n",
    "PHYSICIANS = [\"DR_SMITH\", \"DR_JOHNSON\", \"DR_WILLIAMS\", \"DR_BROWN\", \"DR_JONES\", \"DR_GARCIA\", \"DR_MILLER\", \"DR_DAVIS\"]\n",
    "\n",
    "# Generate patient diagnosis records\n",
    "patient_data = []\n",
    "base_date = datetime(2024, 1, 1)\n",
    "\n",
    "# Create 100 patients with 2-5 diagnoses each\n",
    "for patient_num in range(1, 101):\n",
    "    patient_id = f\"PAT{patient_num:04d}\"\n",
    "    \n",
    "    # Each patient gets 2-5 diagnoses over several months\n",
    "    num_diagnoses = random.randint(2, 5)\n",
    "    \n",
    "    for i in range(num_diagnoses):\n",
    "        # Spread diagnoses over 6 months\n",
    "        days_offset = random.randint(0, 180)\n",
    "        diagnosis_date = base_date + timedelta(days=days_offset)\n",
    "        \n",
    "        # Select random diagnosis\n",
    "        diagnosis_code, description, severity = random.choice(DIAGNOSES)\n",
    "        \n",
    "        # Select random facility and physician\n",
    "        facility = random.choice(FACILITIES)\n",
    "        physician = random.choice(PHYSICIANS)\n",
    "        \n",
    "        patient_data.append({\n",
    "            \"patient_id\": patient_id,\n",
    "            \"diagnosis_date\": diagnosis_date.date(),\n",
    "            \"diagnosis_code\": diagnosis_code,\n",
    "            \"diagnosis_description\": description,\n",
    "            \"severity_level\": severity,\n",
    "            \"treating_physician\": physician,\n",
    "            \"facility_id\": facility\n",
    "        })\n",
    "\n",
    "print(f\"Generated {len(patient_data)} patient diagnosis records\")\n",
    "print(\"Sample record:\", patient_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",
    "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:36:13.458Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame Schema:\n",
       "root\n",
       " |-- patient_id: string (nullable = true)\n",
       " |-- diagnosis_date: date (nullable = true)\n",
       " |-- diagnosis_code: string (nullable = true)\n",
       " |-- diagnosis_description: string (nullable = true)\n",
       " |-- severity_level: string (nullable = true)\n",
       " |-- treating_physician: string (nullable = true)\n",
       " |-- facility_id: string (nullable = true)\n",
       "\n",
       "\n",
       "Sample Data:\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+----------+--------------+--------------+---------------------+--------------+------------------+-----------+\n",
       "|patient_id|diagnosis_date|diagnosis_code|diagnosis_description|severity_level|treating_physician|facility_id|\n",
       "+----------+--------------+--------------+---------------------+--------------+------------------+-----------+\n",
       "|   PAT0001|    2024-04-25|        Z51.11| Encounter for ant...|      Critical|       DR_WILLIAMS|    HOSP002|\n",
       "|   PAT0001|    2024-05-09|         E11.9| Type 2 diabetes m...|        Medium|       DR_WILLIAMS|    HOSP002|\n",
       "|   PAT0001|    2024-06-19|         M79.3| Panniculitis, uns...|           Low|          DR_SMITH|  CLINIC002|\n",
       "|   PAT0001|    2024-04-13|           I10| Essential hyperte...|          High|          DR_BROWN|  CLINIC001|\n",
       "|   PAT0002|    2024-04-10|         M79.3| Panniculitis, uns...|           Low|        DR_JOHNSON|    HOSP001|\n",
       "+----------+--------------+--------------+---------------------+--------------+------------------+-----------+\n",
       "only showing top 5 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "\n",
       "Successfully inserted 349 records into healthcare.gold.patient_diagnoses_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",
    "# Using fully qualified function references to avoid conflicts\n",
    "\n",
    "# Create DataFrame from generated data\n",
    "df_diagnoses = spark.createDataFrame(patient_data, schema=data_schema)\n",
    "\n",
    "# Display schema and sample data\n",
    "print(\"DataFrame Schema:\")\n",
    "df_diagnoses.printSchema()\n",
    "\n",
    "print(\"\\nSample Data:\")\n",
    "df_diagnoses.show(5)\n",
    "\n",
    "# Insert data into Delta table with liquid clustering\n",
    "# The CLUSTER BY (patient_id, diagnosis_date) will automatically optimize the data layout\n",
    "df_diagnoses.write.mode(\"overwrite\").insertInto(\"healthcare.gold.patient_diagnoses_uf\")\n",
    "\n",
    "print(f\"\\nSuccessfully inserted {df_diagnoses.count()} records into healthcare.gold.patient_diagnoses_uf\")\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. **Patient history lookup** (clustered by patient_id)\n",
    "2. **Time-based analysis** (clustered by diagnosis_date)\n",
    "3. **Combined patient + time queries** (optimal for our clustering)\n",
    "\n",
    "### Expected Performance Benefits\n",
    "\n",
    "With liquid clustering, these queries should be significantly faster because:\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:36:29.082Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "=== Query 1: Patient Diagnosis History ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+----------+--------------+--------------+---------------------+--------------+\n",
       "|patient_id|diagnosis_date|diagnosis_code|diagnosis_description|severity_level|\n",
       "+----------+--------------+--------------+---------------------+--------------+\n",
       "|   PAT0001|    2024-04-13|           I10| Essential hyperte...|          High|\n",
       "|   PAT0001|    2024-04-25|        Z51.11| Encounter for ant...|      Critical|\n",
       "|   PAT0001|    2024-05-09|         E11.9| Type 2 diabetes m...|        Medium|\n",
       "|   PAT0001|    2024-06-19|         M79.3| Panniculitis, uns...|           Low|\n",
       "+----------+--------------+--------------+---------------------+--------------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Records found: 4\n",
       "\n",
       "=== Query 2: Recent Critical Diagnoses ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+--------------+----------+--------------+---------------------+------------------+\n",
       "|diagnosis_date|patient_id|diagnosis_code|diagnosis_description|treating_physician|\n",
       "+--------------+----------+--------------+---------------------+------------------+\n",
       "|    2024-06-26|   PAT0007|        Z51.11| Encounter for ant...|          DR_DAVIS|\n",
       "|    2024-06-24|   PAT0097|        Z51.11| Encounter for ant...|         DR_MILLER|\n",
       "|    2024-06-18|   PAT0082|        Z51.11| Encounter for ant...|          DR_JONES|\n",
       "|    2024-06-08|   PAT0014|        Z51.11| Encounter for ant...|          DR_BROWN|\n",
       "|    2024-05-30|   PAT0075|        Z51.11| Encounter for ant...|         DR_GARCIA|\n",
       "|    2024-05-26|   PAT0015|        Z51.11| Encounter for ant...|          DR_SMITH|\n",
       "|    2024-05-24|   PAT0069|        Z51.11| Encounter for ant...|          DR_SMITH|\n",
       "|    2024-04-28|   PAT0027|        Z51.11| Encounter for ant...|         DR_MILLER|\n",
       "|    2024-04-25|   PAT0001|        Z51.11| Encounter for ant...|       DR_WILLIAMS|\n",
       "|    2024-04-25|   PAT0067|        Z51.11| Encounter for ant...|         DR_MILLER|\n",
       "|    2024-04-23|   PAT0065|        Z51.11| Encounter for ant...|          DR_JONES|\n",
       "|    2024-04-19|   PAT0005|        Z51.11| Encounter for ant...|          DR_SMITH|\n",
       "|    2024-04-17|   PAT0035|        Z51.11| Encounter for ant...|         DR_MILLER|\n",
       "|    2024-04-16|   PAT0094|        Z51.11| Encounter for ant...|        DR_JOHNSON|\n",
       "|    2024-04-15|   PAT0002|        Z51.11| Encounter for ant...|         DR_GARCIA|\n",
       "|    2024-04-14|   PAT0008|        Z51.11| Encounter for ant...|         DR_MILLER|\n",
       "|    2024-04-12|   PAT0099|        Z51.11| Encounter for ant...|          DR_BROWN|\n",
       "|    2024-04-10|   PAT0003|        Z51.11| Encounter for ant...|        DR_JOHNSON|\n",
       "|    2024-04-06|   PAT0063|        Z51.11| Encounter for ant...|         DR_MILLER|\n",
       "+--------------+----------+--------------+---------------------+------------------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Critical diagnoses found: 19\n",
       "\n",
       "=== Query 3: Patient Timeline Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+----------+--------------+--------------+--------------+-----------+\n",
       "|patient_id|diagnosis_date|diagnosis_code|severity_level|facility_id|\n",
       "+----------+--------------+--------------+--------------+-----------+\n",
       "|   PAT0011|    2024-06-15|           I10|          High|  CLINIC002|\n",
       "|   PAT0012|    2024-05-14|        I25.10|          High|    HOSP002|\n",
       "|   PAT0012|    2024-05-23|        I25.10|          High|  CLINIC002|\n",
       "|   PAT0012|    2024-06-18|        Z00.00|           Low|  URGENT001|\n",
       "|   PAT0013|    2024-05-17|         N39.0|        Medium|    HOSP001|\n",
       "|   PAT0014|    2024-05-24|        I25.10|          High|  CLINIC002|\n",
       "|   PAT0014|    2024-06-08|         N39.0|        Medium|  CLINIC002|\n",
       "|   PAT0014|    2024-06-08|        Z51.11|      Critical|  URGENT001|\n",
       "|   PAT0015|    2024-04-16|       J45.909|        Medium|  CLINIC002|\n",
       "|   PAT0015|    2024-05-26|        Z51.11|      Critical|  CLINIC002|\n",
       "|   PAT0015|    2024-06-06|       J45.909|        Medium|  CLINIC001|\n",
       "|   PAT0016|    2024-03-03|        Z00.00|           Low|    HOSP001|\n",
       "|   PAT0016|    2024-03-13|        I25.10|          High|  CLINIC001|\n",
       "|   PAT0016|    2024-05-02|         E11.9|        Medium|  CLINIC002|\n",
       "|   PAT0017|    2024-03-10|        I25.10|          High|  URGENT001|\n",
       "|   PAT0017|    2024-03-25|        I25.10|          High|  CLINIC002|\n",
       "|   PAT0017|    2024-05-25|         E11.9|        Medium|  CLINIC001|\n",
       "|   PAT0018|    2024-03-28|        Z00.00|           Low|  URGENT001|\n",
       "|   PAT0018|    2024-05-04|        Z00.00|           Low|    HOSP001|\n",
       "|   PAT0018|    2024-06-06|         M54.5|           Low|  CLINIC002|\n",
       "+----------+--------------+--------------+--------------+-----------+\n",
       "only showing top 20 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Timeline records found: 22\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Demonstrate liquid clustering benefits with optimized queries\n",
    "\n",
    "# Query 1: Patient history - benefits from patient_id clustering\n",
    "print(\"=== Query 1: Patient Diagnosis History ===\")\n",
    "patient_history = spark.sql(\"\"\"\n",
    "SELECT patient_id, diagnosis_date, diagnosis_code, diagnosis_description, severity_level\n",
    "FROM healthcare.gold.patient_diagnoses_uf\n",
    "WHERE patient_id = 'PAT0001'\n",
    "ORDER BY diagnosis_date\n",
    "\"\"\")\n",
    "\n",
    "patient_history.show()\n",
    "print(f\"Records found: {patient_history.count()}\")\n",
    "\n",
    "# Query 2: Time-based analysis - benefits from diagnosis_date clustering\n",
    "print(\"\\n=== Query 2: Recent Critical Diagnoses ===\")\n",
    "recent_critical = spark.sql(\"\"\"\n",
    "SELECT diagnosis_date, patient_id, diagnosis_code, diagnosis_description, treating_physician\n",
    "FROM healthcare.gold.patient_diagnoses_uf\n",
    "WHERE diagnosis_date >= '2024-04-01' AND severity_level = 'Critical'\n",
    "ORDER BY diagnosis_date DESC\n",
    "\"\"\")\n",
    "\n",
    "recent_critical.show()\n",
    "print(f\"Critical diagnoses found: {recent_critical.count()}\")\n",
    "\n",
    "# Query 3: Combined patient + time query - optimal for our clustering strategy\n",
    "print(\"\\n=== Query 3: Patient Timeline Analysis ===\")\n",
    "patient_timeline = spark.sql(\"\"\"\n",
    "SELECT patient_id, diagnosis_date, diagnosis_code, severity_level, facility_id\n",
    "FROM healthcare.gold.patient_diagnoses_uf\n",
    "WHERE patient_id LIKE 'PAT001%' AND diagnosis_date >= '2024-03-01'\n",
    "ORDER BY patient_id, diagnosis_date\n",
    "\"\"\")\n",
    "\n",
    "patient_timeline.show()\n",
    "print(f\"Timeline records found: {patient_timeline.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 healthcare insights possible with this optimized structure.\n",
    "\n",
    "### Key Analytics\n",
    "\n",
    "- **Diagnosis frequency** by type\n",
    "- **Severity distribution** across facilities\n",
    "- **Physician workload** analysis\n",
    "- **Temporal patterns** in diagnoses"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.status.busy": "2025-12-11T19:36:43.715Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "=== Diagnosis Frequency Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+--------------+-------------------------------------------------------------------------------+---------+----------+\n",
       "|diagnosis_code|diagnosis_description                                                          |frequency|percentage|\n",
       "+--------------+-------------------------------------------------------------------------------+---------+----------+\n",
       "|J45.909       |Unspecified asthma, uncomplicated                                              |43       |12.32     |\n",
       "|Z00.00        |Encounter for general adult medical examination without abnormal findings      |42       |12.03     |\n",
       "|I25.10        |Atherosclerotic heart disease of native coronary artery without angina pectoris|41       |11.75     |\n",
       "|M79.3         |Panniculitis, unspecified                                                      |38       |10.89     |\n",
       "|M54.5         |Low back pain                                                                  |37       |10.60     |\n",
       "|E11.9         |Type 2 diabetes mellitus without complications                                 |35       |10.03     |\n",
       "|Z51.11        |Encounter for antineoplastic chemotherapy                                      |33       |9.46      |\n",
       "|N39.0         |Urinary tract infection, site not specified                                    |29       |8.31      |\n",
       "|I10           |Essential hypertension                                                         |28       |8.02      |\n",
       "|F41.9         |Anxiety disorder, unspecified                                                  |23       |6.59      |\n",
       "+--------------+-------------------------------------------------------------------------------+---------+----------+\n",
       "\n",
       "\n",
       "=== Severity Distribution by Facility ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+--------------+-----+\n",
       "|facility_id|severity_level|count|\n",
       "+-----------+--------------+-----+\n",
       "|  CLINIC001|      Critical|    6|\n",
       "|  CLINIC001|          High|   16|\n",
       "|  CLINIC001|           Low|   26|\n",
       "|  CLINIC001|        Medium|   22|\n",
       "|  CLINIC002|      Critical|    5|\n",
       "|  CLINIC002|          High|   19|\n",
       "|  CLINIC002|           Low|   24|\n",
       "|  CLINIC002|        Medium|   26|\n",
       "|    HOSP001|      Critical|    9|\n",
       "|    HOSP001|          High|    9|\n",
       "|    HOSP001|           Low|   23|\n",
       "|    HOSP001|        Medium|   25|\n",
       "|    HOSP002|      Critical|    7|\n",
       "|    HOSP002|          High|   11|\n",
       "|    HOSP002|           Low|   24|\n",
       "|    HOSP002|        Medium|   25|\n",
       "|  URGENT001|      Critical|    6|\n",
       "|  URGENT001|          High|   14|\n",
       "|  URGENT001|           Low|   20|\n",
       "|  URGENT001|        Medium|   32|\n",
       "+-----------+--------------+-----+\n",
       "\n",
       "\n",
       "=== Physician Workload Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+------------------+---------------+---------------+-------------------+\n",
       "|treating_physician|total_diagnoses|unique_patients|critical_case_ratio|\n",
       "+------------------+---------------+---------------+-------------------+\n",
       "|       DR_WILLIAMS|             51|             39|              0.078|\n",
       "|          DR_BROWN|             49|             41|              0.082|\n",
       "|          DR_SMITH|             47|             37|              0.149|\n",
       "|        DR_JOHNSON|             44|             35|              0.091|\n",
       "|         DR_MILLER|             43|             38|               0.14|\n",
       "|          DR_DAVIS|             41|             37|              0.024|\n",
       "|          DR_JONES|             38|             32|              0.079|\n",
       "|         DR_GARCIA|             36|             33|              0.111|\n",
       "+------------------+---------------+---------------+-------------------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Analyze clustering effectiveness and healthcare insights\n",
    "\n",
    "# Diagnosis frequency analysis\n",
    "print(\"=== Diagnosis Frequency Analysis ===\")\n",
    "diagnosis_freq = spark.sql(\"\"\"\n",
    "SELECT diagnosis_code, diagnosis_description, COUNT(*) as frequency,\n",
    "       ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) as percentage\n",
    "FROM healthcare.gold.patient_diagnoses_uf\n",
    "GROUP BY diagnosis_code, diagnosis_description\n",
    "ORDER BY frequency DESC\n",
    "\"\"\")\n",
    "\n",
    "diagnosis_freq.show(truncate=False)\n",
    "\n",
    "# Severity distribution by facility\n",
    "print(\"\\n=== Severity Distribution by Facility ===\")\n",
    "severity_by_facility = spark.sql(\"\"\"\n",
    "SELECT facility_id, severity_level, COUNT(*) as count\n",
    "FROM healthcare.gold.patient_diagnoses_uf\n",
    "GROUP BY facility_id, severity_level\n",
    "ORDER BY facility_id, severity_level\n",
    "\"\"\")\n",
    "\n",
    "severity_by_facility.show()\n",
    "\n",
    "# Physician workload analysis\n",
    "print(\"\\n=== Physician Workload Analysis ===\")\n",
    "physician_workload = spark.sql(\"\"\"\n",
    "SELECT treating_physician, COUNT(*) as total_diagnoses,\n",
    "       COUNT(DISTINCT patient_id) as unique_patients,\n",
    "       ROUND(AVG(CASE WHEN severity_level = 'Critical' THEN 1 ELSE 0 END), 3) as critical_case_ratio\n",
    "FROM healthcare.gold.patient_diagnoses_uf\n",
    "GROUP BY treating_physician\n",
    "ORDER BY total_diagnoses DESC\n",
    "\"\"\")\n",
    "\n",
    "physician_workload.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Key Takeaways: Iceberg and Liquid Clustering in AIDP\n",
    "\n",
    "### What We Demonstrated\n",
    "\n",
    "1. **Iceberg Compatibility**: Enabled Delta Universal Format with `'delta.universalFormat.enabledFormats' = 'iceberg'` for cross-engine access\n",
    "\n",
    "2. **Liquid Clustering**: Created a table with `CLUSTER BY (patient_id, diagnosis_date)` for automatic data optimization\n",
    "\n",
    "3. **Performance Benefits**: Queries on clustered columns are significantly faster due to data locality\n",
    "\n",
    "4. **Zero Maintenance**: No manual partitioning, bucketing, or Z-Ordering required\n",
    "\n",
    "5. **Real-World Use Case**: Healthcare analytics where patient history lookups and temporal analysis are critical\n",
    "\n",
    "### Iceberg Advantages\n",
    "\n",
    "- **Open Standard**: Apache 2.0 licensed, community-driven table format\n",
    "- **Schema Evolution**: Add, drop, rename columns without expensive data rewrites\n",
    "- **Partition Evolution**: Change partitioning schemes without disrupting workflows\n",
    "- **Time Travel**: Query historical data snapshots for auditing and reproducibility\n",
    "- **ACID Transactions**: Reliable concurrent read/write operations across engines\n",
    "- **Multi-Engine Support**: Query same data from Spark, Presto, Flink, Hive, and more\n",
    "- **Future-Proof**: Standards-based approach protects your data investments\n",
    "\n",
    "### AIDP Advantages\n",
    "\n",
    "- **Unified Analytics**: Seamlessly integrates with other AIDP services\n",
    "- **Governance**: Catalog and schema isolation for healthcare data\n",
    "- **Performance**: Optimized for both OLAP and OLTP workloads\n",
    "- **Scalability**: Handles healthcare-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. **Leverage Iceberg features** like schema evolution for changing requirements\n",
    "5. **Monitor and adjust** as query patterns and schema evolve\n",
    "\n",
    "### Next Steps\n",
    "\n",
    "- Explore Iceberg time travel capabilities with `SELECT * FROM table TIMESTAMP AS OF`\n",
    "- Try schema evolution by adding new columns without data migration\n",
    "- Query the same data from different engines like Presto or Trino\n",
    "- Integrate with real healthcare systems\n",
    "- Scale up to larger healthcare datasets across multiple clusters\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": 1,
  "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
}