{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "## Oracle AI Data Platform v1.0\n", "\n", "Copyright © 2026, Oracle and/or its affiliates.\n", "\n", "Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Apache Iceberg Table with OCI Native Protocol\n", "\n", "This notebook demonstrates how to create and interact with an **Apache Iceberg** table stored in **OCI Object Storage**, using the native `oci://` protocol via a **Hadoop Catalog**.\n", "\n", "---\n", "\n", "## What is Apache Iceberg?\n", "\n", "Apache Iceberg is an open **table format** for large-scale analytic datasets. It is not a file format (like Parquet or Avro) — it sits on top of file formats and adds powerful features that raw files alone cannot provide such as Schema evolution, Time travel, ACID transactions, just to name a few.\n", "\n", "Iceberg stores its own **metadata layer** alongside the data files. This metadata tracks snapshots, schema versions, partition layout, and data file locations.\n", "\n", "---\n", "\n", "## What is a Hadoop Catalog?\n", "\n", "Iceberg requires a **catalog** to track where tables are and what their current snapshot is. There are several catalog types:\n", "\n", "- **Hadoop Catalog**: Stores metadata directly in the filesystem (Object Storage). Simple, no external service needed. This is what this notebook uses.\n", "- **Hive Metastore**: Uses an external Hive Metastore service for catalog management.\n", "- **REST Catalog**: Uses a REST API backed by a catalog server. Enables centralized governance, fine-grained access control, and multi-engine interoperability.\n", "- **Master Catalog**: The built-in catalog provided by OCI AI Data Platform. Fully managed, integrated with OCI security, and the recommended option for production workloads on the platform.\n", "\n", "This notebook uses the **Hadoop Catalog** — metadata is written directly into the OCI bucket alongside the data. This is the simplest setup and works great for getting started or for simple architectures.\n", "\n", "---\n", "\n", "## Why OCI Native Protocol (`oci://`)?\n", "\n", "OCI Object Storage can be accessed via two protocols in AI Data Platform (running Spark enabled clusters):\n", "\n", "- **S3 Compatible API** (`s3a://`): Requires S3A JARs, endpoint configuration, and credentials setup.\n", "- **OCI Native (`oci://`)**: Works out of the box in the AI Data Platform Workbench. Performs authentication automatically based on your user or group permissions.\n", "\n", "This notebook uses `oci://` — no extra JARs or credential setup needed.\n", "\n", "Note that all metadata in the catalog will reference `oci://` paths, not `s3a://`.\n", "\n", "---\n", "\n", "## Prerequisites\n", "\n", "- Running in OCI AI Data Platform Workbench\n", "- An OCI Object Storage bucket already created\n", "- A folder (prefix) created inside the bucket to serve as the Iceberg warehouse\n", "- Permissions already set on OCI tenancy for that bucket.Example of statements: \n", " - `Allow dynamic-group to manage objects in compartment where target.bucket.name=''`\n", " - `Allow dynamic-group to manage buckets in compartment where target.bucket.name=''`\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 1: Verify Spark Session\n", "\n", "In AI Data Platform Workbench, a Spark session (`spark`) is automatically available when running on a Spark-enabled cluster. Let's confirm it's active before proceeding." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print(f\"Spark version: {spark.version}\")\n", "print(f\"Application ID: {spark.sparkContext.applicationId}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 2: Configuration\n", "\n", "Fill in the variables below with your OCI environment details.\n", "\n", "> **Tip:** The warehouse path follows the format `oci://bucket@namespace/folder`. The namespace is your OCI tenancy's Object Storage namespace, visible in the OCI Console under Object Storage settings." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# ── OCI Object Storage ────────────────────────────────────────────────────────\n", "# Your tenancy's Object Storage namespace (found in OCI Console > Object Storage)\n", "OCI_NAMESPACE = \"\"\n", "\n", "# Region where your bucket is located\n", "OCI_REGION = \"us-ashburn-1\"\n", "\n", "# Name of the OCI bucket where Iceberg data will be stored\n", "BUCKET_NAME = \"\"\n", "\n", "# ── Iceberg Warehouse ─────────────────────────────────────────────────────────\n", "# The warehouse is the root folder inside the bucket where Iceberg will store\n", "# all its databases, tables, data files, and metadata.\n", "WAREHOUSE_NAME = \"iceberg-warehouse\"\n", "\n", "# Iceberg warehouse path using OCI native protocol: oci://bucket@namespace/path\n", "WAREHOUSE_PATH = f\"oci://{BUCKET_NAME}@{OCI_NAMESPACE}/{WAREHOUSE_NAME}\"\n", "\n", "# ── Catalog, Database, and Table names ────────────────────────────────────────\n", "# The catalog is the top-level namespace registered in Spark.\n", "# In Iceberg SQL, tables are always referenced as: catalog.database.table\n", "CATALOG_NAME = \"oci_catalog\"\n", "\n", "# A database (also called namespace) groups related tables together.\n", "DATABASE_NAME = \"\"\n", "\n", "# The table name\n", "TABLE_NAME = \"\"\n", "\n", "# Fully qualified table name used in all SQL statements\n", "FULL_TABLE_NAME = f\"{CATALOG_NAME}.{DATABASE_NAME}.{TABLE_NAME}\"\n", "\n", "print(f\"Warehouse : {WAREHOUSE_PATH}\")\n", "print(f\"Table : {FULL_TABLE_NAME}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 3: Configure the Iceberg Catalog in Spark\n", "\n", "Spark does not know about Iceberg catalogs by default. We need to register our Hadoop Catalog by setting Spark configuration properties at runtime.\n", "\n", "These three properties are the minimum required:\n", "- `spark.sql.catalog.`: the Iceberg catalog implementation class\n", "- `spark.sql.catalog..type`: catalog type (`hadoop`, `hive`, or `rest`)\n", "- `spark.sql.catalog..warehouse`: the root path for all table data and metadata" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Register the Iceberg catalog in Spark using the Hadoop catalog type.\n", "# Authentication to OCI Object Storage is handled automatically based on your Workbench permissions.\n", "spark.conf.set(f\"spark.sql.catalog.{CATALOG_NAME}\", \"org.apache.iceberg.spark.SparkCatalog\")\n", "spark.conf.set(f\"spark.sql.catalog.{CATALOG_NAME}.type\", \"hadoop\")\n", "spark.conf.set(f\"spark.sql.catalog.{CATALOG_NAME}.warehouse\", WAREHOUSE_PATH)\n", "\n", "print(f\"Catalog '{CATALOG_NAME}' registered\")\n", "print(f\"Type : Hadoop (metadata stored in Object Storage)\")\n", "print(f\"Warehouse: {WAREHOUSE_PATH}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 4: Create a Database\n", "\n", "In Iceberg, a **database** (also called a namespace) is a logical grouping of tables — similar to a schema in traditional databases.\n", "\n", "With a Hadoop Catalog, creating a database simply creates a folder under the warehouse path:\n", "```\n", "oci://bucket@namespace///\n", "```" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "spark.sql(f\"CREATE DATABASE IF NOT EXISTS {CATALOG_NAME}.{DATABASE_NAME}\")\n", "\n", "print(f\"Database '{DATABASE_NAME}' is ready\")\n", "print(f\"\\nAll databases in catalog '{CATALOG_NAME}':\")\n", "spark.sql(f\"SHOW DATABASES IN {CATALOG_NAME}\").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 5: Create an Iceberg Table\n", "\n", "When creating an Iceberg table you define:\n", "- **Schema**: column names and types\n", "- **File format**: `USING iceberg` tells Spark this is an Iceberg table (data files will be Parquet by default)\n", "- **Partitioning** *(optional)*: Iceberg supports hidden partitioning — you declare the partition column and Iceberg handles the physical layout automatically, without exposing partition columns to query writers\n", "\n", "> **Note on `DROP TABLE IF EXISTS`:** The line below drops the table if it already exists so this notebook can be safely re-run. In production, remove that line to avoid accidental data loss." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Drop the table if it already exists — safe for re-running this notebook.\n", "# Remove this line in production to avoid accidental data loss.\n", "spark.sql(f\"DROP TABLE IF EXISTS {FULL_TABLE_NAME}\")\n", "\n", "spark.sql(f\"\"\"\n", " CREATE TABLE {FULL_TABLE_NAME} (\n", " employee_id INT,\n", " employee_name STRING,\n", " salary DOUBLE,\n", " department STRING,\n", " hire_date DATE\n", " )\n", " USING iceberg\n", " PARTITIONED BY (department)\n", "\"\"\")\n", "\n", "# Iceberg stores data as Parquet files by default.\n", "# The 'department' partition means each department gets its own subfolder,\n", "# so queries filtering by department skip irrelevant files entirely (partition pruning).\n", "print(f\"Table created: {FULL_TABLE_NAME}\")\n", "print(f\"Location : {WAREHOUSE_PATH}/{DATABASE_NAME}/{TABLE_NAME}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 6: Insert Sample Data\n", "\n", "Instead of hardcoding values directly in SQL, we build a **Pandas DataFrame** first and convert it to a Spark DataFrame before writing. This pattern is more reusable — in practice you would replace the Pandas DataFrame with data loaded from a file, an API, a database, or any other source." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "from datetime import date\n", "\n", "# Build your data as a Pandas DataFrame.\n", "# Replace this with any data source: pd.read_csv(), pd.read_json(), API response, etc.\n", "df = pd.DataFrame([\n", " (101, \"John Doe\", 75000.0, \"Engineering\", date(2022, 1, 15)),\n", " (102, \"Jane Smith\", 85000.0, \"Sales\", date(2021, 3, 20)),\n", " (103, \"Bob Johnson\", 65000.0, \"Engineering\", date(2023, 6, 10)),\n", " (104, \"Alice Williams\", 95000.0, \"Management\", date(2020, 8, 5)),\n", " (105, \"Charlie Brown\", 70000.0, \"HR\", date(2022, 11, 30)),\n", " (106, \"Diana Prince\", 88000.0, \"Engineering\", date(2021, 9, 12)),\n", " (107, \"Eve Adams\", 72000.0, \"Sales\", date(2023, 2, 18)),\n", "], columns=[\"employee_id\", \"employee_name\", \"salary\", \"department\", \"hire_date\"])\n", "\n", "# Convert to a Spark DataFrame and write to the Iceberg table.\n", "# Each writeTo() call is a full ACID transaction and creates a new Iceberg snapshot.\n", "spark.createDataFrame(df).writeTo(FULL_TABLE_NAME).append()\n", "\n", "print(f\"{len(df)} rows written — this created snapshot #1\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 7: Query the Table\n", "\n", "Standard SQL works as expected. Because the table is partitioned by `department`, any `WHERE department = '...'` filter will only scan the relevant partition files — this is called **partition pruning**." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Full table scan\n", "print(\"All employees:\")\n", "spark.sql(f\"SELECT * FROM {FULL_TABLE_NAME} ORDER BY department, employee_id\").show()\n", "\n", "# Filtered query — only scans the Engineering partition files\n", "print(\"Engineering department (partition pruning in action):\")\n", "spark.sql(f\"\"\"\n", " SELECT employee_id, employee_name, salary\n", " FROM {FULL_TABLE_NAME}\n", " WHERE department = 'Engineering'\n", " ORDER BY salary DESC\n", "\"\"\").show()\n", "\n", "# Aggregation by department\n", "print(\"Department summary:\")\n", "spark.sql(f\"\"\"\n", " SELECT department,\n", " COUNT(*) AS headcount,\n", " ROUND(AVG(salary), 2) AS avg_salary,\n", " MIN(hire_date) AS earliest_hire\n", " FROM {FULL_TABLE_NAME}\n", " GROUP BY department\n", " ORDER BY avg_salary DESC\n", "\"\"\").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 8: Schema Evolution\n", "\n", "One of Iceberg's most important features is **safe schema evolution**. You can add, rename, or drop columns without rewriting existing data files.\n", "\n", "Iceberg tracks schema changes in its metadata. When old files are read after a column is added, the missing column simply returns `null` (or its default value). This is safe and transparent to query writers." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Add a new column — no data rewrite needed\n", "spark.sql(f\"ALTER TABLE {FULL_TABLE_NAME} ADD COLUMN location STRING\")\n", "\n", "# Insert new data that includes the new column\n", "spark.sql(f\"\"\"\n", " INSERT INTO {FULL_TABLE_NAME} VALUES\n", " (108, 'Frank Castle', 80000.0, 'Engineering', DATE '2024-01-10', 'New York'),\n", " (109, 'Grace Hopper', 91000.0, 'Management', DATE '2019-05-22', 'Boston')\n", "\"\"\")\n", "\n", "# When reading, existing rows will show NULL for 'location' — no errors, no data loss\n", "print(\"Table after schema evolution (location is NULL for old rows):\")\n", "spark.sql(f\"SELECT * FROM {FULL_TABLE_NAME} ORDER BY employee_id\").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 9: Snapshots and Time Travel\n", "\n", "Every write operation in Iceberg (INSERT, UPDATE, DELETE, schema change) creates a new **snapshot**. A snapshot is an immutable, point-in-time view of the entire table.\n", "\n", "This enables **time travel**: querying the table as it was at a specific snapshot or timestamp, without any additional setup." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# List all snapshots — each write created one\n", "snapshots_df = spark.sql(f\"\"\"\n", " SELECT snapshot_id, committed_at, operation, summary\n", " FROM {FULL_TABLE_NAME}.snapshots\n", " ORDER BY committed_at\n", "\"\"\")\n", "snapshots_df.show(truncate=False)\n", "\n", "# Get the first snapshot ID (before schema evolution and the new rows)\n", "first_snapshot_id = snapshots_df.collect()[0][\"snapshot_id\"]\n", "print(f\"First snapshot ID: {first_snapshot_id}\")\n", "\n", "# Time travel: query the table as it was at the first snapshot\n", "# At that point it had only 7 rows and no 'location' column\n", "print(\"\\nTable at first snapshot (7 rows, original schema):\")\n", "spark.sql(f\"\"\"\n", " SELECT * FROM {FULL_TABLE_NAME}\n", " VERSION AS OF {first_snapshot_id}\n", " ORDER BY employee_id\n", "\"\"\").show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Step 10: Inspect Physical Files\n", "\n", "Iceberg exposes metadata tables you can query like regular tables. The `.files` metadata table shows the actual Parquet data files that back the current snapshot.\n", "\n", "Notice how files are organized by partition — each department gets its own directory and file(s)." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "files_df = spark.sql(f\"\"\"\n", " SELECT file_path, file_format, record_count, file_size_in_bytes\n", " FROM {FULL_TABLE_NAME}.files\n", "\"\"\")\n", "\n", "print(\"Physical data files in OCI Object Storage:\")\n", "files_df.show(truncate=False)\n", "\n", "# Summary stats\n", "total_files = files_df.count()\n", "total_records = files_df.agg({\"record_count\": \"sum\"}).collect()[0][0]\n", "total_size_kb = files_df.agg({\"file_size_in_bytes\": \"sum\"}).collect()[0][0] / 1024\n", "\n", "print(f\"Files : {total_files} (one per partition per write batch)\")\n", "print(f\"Records: {total_records}\")\n", "print(f\"Size : {total_size_kb:.1f} KB\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary\n", "\n", "This notebook covered the end-to-end workflow for creating and using an Apache Iceberg table in OCI Object Storage with the native `oci://` protocol.\n", "\n", "### What was demonstrated\n", "\n", "| Step | Concept |\n", "|---|---|\n", "| Catalog registration | Registering a Hadoop Catalog in Spark at runtime |\n", "| Database creation | Logical grouping of Iceberg tables |\n", "| Table creation | Schema definition with partitioning |\n", "| Insert | ACID transactions and snapshot creation |\n", "| Querying | Standard SQL with partition pruning |\n", "| Schema evolution | Adding columns without rewriting data |\n", "| Time travel | Querying past snapshots with `VERSION AS OF` |\n", "| Metadata inspection | Using `.files` and `.snapshots` metadata tables |\n", "\n", "### Key takeaways\n", "\n", "- **No S3A JARs needed**: `oci://` works natively in OCI environments\n", "- **Authentication is automatic**: permissions are resolved automatically in AI Data Platform Workbench\n", "- **Iceberg is a table format, not a file format**: data files are Parquet, Iceberg manages the metadata layer on top\n", "- **Schema evolution is safe**: add or drop columns without breaking existing readers or rewriting files\n", "- **Every write is a snapshot**: time travel comes for free\n", "\n", "---" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.14.3" } }, "nbformat": 4, "nbformat_minor": 4 }