iceberg_pyspark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reference Material\n",
    "\n",
    "[EMR Iceberg Documentation](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-iceberg.html)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Configuration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before running the code in the cell(s) below double check that the EMR cluster attached to this notebook was created with the following in it's software settings\n",
    "\n",
    "```[{\"classification\":\"iceberg-defaults\",\"properties\":{\"iceberg.enabled\":true}}]```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Update the S3 path in the ``` \"spark.sql.catalog.dev.warehouse\": ``` part of the configuration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:00:22.186923Z",
     "iopub.status.busy": "2022-10-08T18:00:22.186686Z",
     "iopub.status.idle": "2022-10-08T18:00:22.202762Z",
     "shell.execute_reply": "2022-10-08T18:00:22.202086Z",
     "shell.execute_reply.started": "2022-10-08T18:00:22.186895Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "Current session configs: <tt>{'conf': {'spark.sql.extensions': 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions', 'spark.sql.catalog.dev': 'org.apache.iceberg.spark.SparkCatalog', 'spark.sql.catalog.dev.type': 'hadoop', 'spark.sql.catalog.dev.warehouse': 's3://emr-studio-demo-s3bucket-g3frvpeeuanh/iceberg/'}, 'proxyUser': 'assumed-role_user-role-EMR-studio_ChrisSharkey', 'kind': 'pyspark'}</tt><br>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "No active sessions."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%configure -f\n",
    "{\n",
    "\"conf\":{\n",
    "    \"spark.sql.extensions\":\"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions\",\n",
    "    \"spark.sql.catalog.dev\":\"org.apache.iceberg.spark.SparkCatalog\",\n",
    "    \"spark.sql.catalog.dev.type\":\"hadoop\",\n",
    "    \"spark.sql.catalog.dev.warehouse\":\"s3://emr-studio-demo-s3bucket-g3frvpeeuanh/iceberg/\"\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set a variable equal to the name of the S3 bucket to read / write from"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:00:23.923073Z",
     "iopub.status.busy": "2022-10-08T18:00:23.922843Z",
     "iopub.status.idle": "2022-10-08T18:00:57.657773Z",
     "shell.execute_reply": "2022-10-08T18:00:57.656846Z",
     "shell.execute_reply.started": "2022-10-08T18:00:23.923048Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "1c3197795d4d4aee881b4a1c3d58d651",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n<tbody><tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>User</th><th>Current session?</th></tr><tr><td>14</td><td>application_1665227735286_0015</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-0-0-5.ec2.internal:20888/proxy/application_1665227735286_0015/\" class=\"emr-proxy-link j-18GB9AZ4QUV6P application_1665227735286_0015\" emr-resource=\"j-18GB9AZ4QUV6P\n\" application-id=\"application_1665227735286_0015\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-0-0-5.ec2.internal:8042/node/containerlogs/container_1665227735286_0015_01_000001/livy\">Link</a></td><td>None</td><td>✔</td></tr></tbody></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "s3_bucket_name = \"emr-studio-demo-s3bucket-g3frvpeeuanh\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:00:57.661765Z",
     "iopub.status.busy": "2022-10-08T18:00:57.661191Z",
     "iopub.status.idle": "2022-10-08T18:00:57.804558Z",
     "shell.execute_reply": "2022-10-08T18:00:57.803616Z",
     "shell.execute_reply.started": "2022-10-08T18:00:57.661722Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "dfe30a3dc7e44bd3ad5bcb6210ad43b8",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create an Iceberg Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:00:57.807139Z",
     "iopub.status.busy": "2022-10-08T18:00:57.806665Z",
     "iopub.status.idle": "2022-10-08T18:00:57.961137Z",
     "shell.execute_reply": "2022-10-08T18:00:57.960322Z",
     "shell.execute_reply.started": "2022-10-08T18:00:57.807099Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "dbfdcf41e1b34777baf2f737cf16d8eb",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Drop the table if you need to ...\n",
    "# spark.sql(\" DROP TABLE dev.db.iceberg_table \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:00:57.963011Z",
     "iopub.status.busy": "2022-10-08T18:00:57.962543Z",
     "iopub.status.idle": "2022-10-08T18:01:05.336718Z",
     "shell.execute_reply": "2022-10-08T18:01:05.335937Z",
     "shell.execute_reply.started": "2022-10-08T18:00:57.962971Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "94e652950b4c4c26a25e3c94d390801b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DataFrame[]"
     ]
    }
   ],
   "source": [
    "# Create an Iceberg table\n",
    "spark.sql(\" CREATE TABLE IF NOT EXISTS dev.db.iceberg_table (id string, name string, create_date string, last_update_time string) USING iceberg LOCATION 's3://\" + s3_bucket_name + \"/iceberg/db/iceberg_table' \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:05.337964Z",
     "iopub.status.busy": "2022-10-08T18:01:05.337783Z",
     "iopub.status.idle": "2022-10-08T18:01:06.661105Z",
     "shell.execute_reply": "2022-10-08T18:01:06.660092Z",
     "shell.execute_reply.started": "2022-10-08T18:01:05.337941Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "803d50917e7f49b59b3094e5ffedd558",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|namespace|\n",
      "+---------+\n",
      "|       db|\n",
      "+---------+"
     ]
    }
   ],
   "source": [
    "# List all spark SQL databases\n",
    "spark.sql(\" SHOW DATABASES in dev \").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:06.662780Z",
     "iopub.status.busy": "2022-10-08T18:01:06.662544Z",
     "iopub.status.idle": "2022-10-08T18:01:10.031753Z",
     "shell.execute_reply": "2022-10-08T18:01:10.030999Z",
     "shell.execute_reply.started": "2022-10-08T18:01:06.662748Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "6d30eb4a1c594696adecf9f45d998eeb",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-------------+-----------+\n",
      "|namespace|    tableName|isTemporary|\n",
      "+---------+-------------+-----------+\n",
      "|       db|iceberg_table|      false|\n",
      "+---------+-------------+-----------+"
     ]
    }
   ],
   "source": [
    "# List all spark SQL tables\n",
    "spark.sql(\" SHOW TABLES IN dev.db \").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:16.043245Z",
     "iopub.status.busy": "2022-10-08T18:01:16.043017Z",
     "iopub.status.idle": "2022-10-08T18:01:16.309607Z",
     "shell.execute_reply": "2022-10-08T18:01:16.308857Z",
     "shell.execute_reply.started": "2022-10-08T18:01:16.043220Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "6068f575f6434c469533d0bc083cdb53",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "data = [\n",
    "        (1, \"Chris\", \"2020-01-01\", datetime.strptime('2020-01-01 00:00:00', '%Y-%m-%d %H:%M:%S')),\n",
    "        (2, \"Will\", \"2020-01-01\", datetime.strptime('2020-01-01 00:00:00', '%Y-%m-%d %H:%M:%S')),\n",
    "        (3, \"Emma\", \"2020-01-01\", datetime.strptime('2020-01-01 00:00:00', '%Y-%m-%d %H:%M:%S')),\n",
    "        (4, \"John\", \"2020-01-01\", datetime.strptime('2020-01-01 00:00:00', '%Y-%m-%d %H:%M:%S')),\n",
    "        (5, \"Eric\", \"2020-01-01\", datetime.strptime('2020-01-01 00:00:00', '%Y-%m-%d %H:%M:%S')),\n",
    "        (6, \"Adam\", \"2020-01-01\", datetime.strptime('2020-01-01 00:00:00', '%Y-%m-%d %H:%M:%S'))\n",
    "]\n",
    "\n",
    "schema = StructType([\n",
    "        StructField(\"id\", IntegerType(), False),\n",
    "        StructField(\"name\", StringType(), False), \n",
    "        StructField(\"create_date\", StringType(), False),             \n",
    "        StructField(\"last_update_time\", TimestampType(), False)    \n",
    "])\n",
    "\n",
    "inputDF = spark.createDataFrame(data=data,schema=schema)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Write Data to Iceberg Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:18.474067Z",
     "iopub.status.busy": "2022-10-08T18:01:18.473820Z",
     "iopub.status.idle": "2022-10-08T18:01:23.795325Z",
     "shell.execute_reply": "2022-10-08T18:01:23.794352Z",
     "shell.execute_reply.started": "2022-10-08T18:01:18.474038Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "99f4e49bf23e4154a6fdc8ac9788f39f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "inputDF.writeTo(\"dev.db.iceberg_table\").append()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read Data from Iceberg Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:23.797319Z",
     "iopub.status.busy": "2022-10-08T18:01:23.797069Z",
     "iopub.status.idle": "2022-10-08T18:01:26.102291Z",
     "shell.execute_reply": "2022-10-08T18:01:26.101597Z",
     "shell.execute_reply.started": "2022-10-08T18:01:23.797284Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "26483c257c9146f2a0bdc22011844b5b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+-----------+-------------------+\n",
      "| id| name|create_date|   last_update_time|\n",
      "+---+-----+-----------+-------------------+\n",
      "|  1|Chris| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  2| Will| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  3| Emma| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  4| John| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  5| Eric| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  6| Adam| 2020-01-01|2020-01-01 00:00:00|\n",
      "+---+-----+-----------+-------------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\" SELECT * FROM dev.db.iceberg_table \").show() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Update and Delete Data to Iceberg Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:30.242705Z",
     "iopub.status.busy": "2022-10-08T18:01:30.242456Z",
     "iopub.status.idle": "2022-10-08T18:01:30.509495Z",
     "shell.execute_reply": "2022-10-08T18:01:30.508850Z",
     "shell.execute_reply.started": "2022-10-08T18:01:30.242681Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c6aca906f19b426d8bf6e5cd55236265",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "data = [\n",
    "        (1, \"Christopher\", \"2020-01-01\", datetime.strptime('2020-01-02 00:00:00', '%Y-%m-%d %H:%M:%S'), \"update\"),\n",
    "        (3, \"Emmeline\", \"2020-01-01\", datetime.strptime('2020-01-02 00:00:00', '%Y-%m-%d %H:%M:%S'), \"update\"),\n",
    "        (5, \"Eric\", \"2020-01-01\", datetime.strptime('2020-01-02 00:00:00', '%Y-%m-%d %H:%M:%S'), \"delete\"),\n",
    "        (7, \"Bill\", \"2020-01-02\", datetime.strptime('2020-01-02 00:00:00', '%Y-%m-%d %H:%M:%S'), \"append\")\n",
    "]\n",
    "\n",
    "schema = StructType([\n",
    "        StructField(\"id\", IntegerType(), False),\n",
    "        StructField(\"name\", StringType(), False), \n",
    "        StructField(\"create_date\", StringType(), False),             \n",
    "        StructField(\"last_update_time\", TimestampType(), False),\n",
    "        StructField(\"change_type\", StringType(), False)\n",
    "])\n",
    "\n",
    "mergeDF = spark.createDataFrame(data=data,schema=schema)\n",
    "\n",
    "mergeDF.createOrReplaceTempView(\"mergeTable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:32.252410Z",
     "iopub.status.busy": "2022-10-08T18:01:32.252171Z",
     "iopub.status.idle": "2022-10-08T18:01:32.314293Z",
     "shell.execute_reply": "2022-10-08T18:01:32.313611Z",
     "shell.execute_reply.started": "2022-10-08T18:01:32.252383Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ff39ededa0744baa8e6c91c412e36351",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# spark.sql(\" DROP TABLE mergeTable \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:01:56.080416Z",
     "iopub.status.busy": "2022-10-08T18:01:56.080174Z",
     "iopub.status.idle": "2022-10-08T18:02:05.427223Z",
     "shell.execute_reply": "2022-10-08T18:02:05.426465Z",
     "shell.execute_reply.started": "2022-10-08T18:01:56.080391Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "37c996480291446b82f9bce4bbaa394b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DataFrame[]"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\" \n",
    "    MERGE INTO \n",
    "        dev.db.iceberg_table t \n",
    "    USING \n",
    "        (SELECT * FROM mergeTable) s \n",
    "    ON \n",
    "        t.id = s.id\n",
    "    WHEN MATCHED AND s.change_type = 'update' THEN UPDATE SET t.name = s.name, t.last_update_time = s.last_update_time \n",
    "    WHEN MATCHED AND s.change_type = 'delete' THEN DELETE\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:02:05.428967Z",
     "iopub.status.busy": "2022-10-08T18:02:05.428713Z",
     "iopub.status.idle": "2022-10-08T18:02:06.215043Z",
     "shell.execute_reply": "2022-10-08T18:02:06.213862Z",
     "shell.execute_reply.started": "2022-10-08T18:02:05.428931Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "157a8a3e5ff94680b54cd49664e3fc46",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+-----------+-------------------+\n",
      "| id|       name|create_date|   last_update_time|\n",
      "+---+-----------+-----------+-------------------+\n",
      "|  1|Christopher| 2020-01-01|2020-01-02 00:00:00|\n",
      "|  2|       Will| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  3|   Emmeline| 2020-01-01|2020-01-02 00:00:00|\n",
      "|  4|       John| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  6|       Adam| 2020-01-01|2020-01-01 00:00:00|\n",
      "+---+-----------+-----------+-------------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\" SELECT * FROM dev.db.iceberg_table ORDER BY id \").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Snapshots"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:02:21.535838Z",
     "iopub.status.busy": "2022-10-08T18:02:21.535604Z",
     "iopub.status.idle": "2022-10-08T18:02:22.307863Z",
     "shell.execute_reply": "2022-10-08T18:02:22.307162Z",
     "shell.execute_reply.started": "2022-10-08T18:02:21.535813Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a491a6088e844751b13ad70e6181dd4f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------------+-------------------+---------+--------------------+--------------------+\n",
      "|        committed_at|        snapshot_id|          parent_id|operation|       manifest_list|             summary|\n",
      "+--------------------+-------------------+-------------------+---------+--------------------+--------------------+\n",
      "|2022-10-08 18:01:...|4202173763955982227|               null|   append|s3://emr-studio-d...|{spark.app.id -> ...|\n",
      "|2022-10-08 18:02:...|2623371660901291398|4202173763955982227|overwrite|s3://emr-studio-d...|{spark.app.id -> ...|\n",
      "+--------------------+-------------------+-------------------+---------+--------------------+--------------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\" SELECT * FROM dev.db.iceberg_table.snapshots \").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:02:22.309615Z",
     "iopub.status.busy": "2022-10-08T18:02:22.309352Z",
     "iopub.status.idle": "2022-10-08T18:02:24.592746Z",
     "shell.execute_reply": "2022-10-08T18:02:24.590502Z",
     "shell.execute_reply.started": "2022-10-08T18:02:22.309577Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "5091a7cffeff40ad95aead73f49a9d93",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Add a new record\n",
    "data = [\n",
    "        (7, \"Bill\", \"2020-01-02\", datetime.strptime('2020-01-03 00:00:00', '%Y-%m-%d %H:%M:%S'))\n",
    "]\n",
    "\n",
    "schema = StructType([\n",
    "        StructField(\"id\", IntegerType(), False),\n",
    "        StructField(\"name\", StringType(), False), \n",
    "        StructField(\"create_date\", StringType(), False),             \n",
    "        StructField(\"last_update_time\", TimestampType(), False)    \n",
    "])\n",
    "\n",
    "appendDF = spark.createDataFrame(data=data,schema=schema)\n",
    "\n",
    "appendDF.writeTo(\"dev.db.iceberg_table\").append()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Time Travel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:09:21.187225Z",
     "iopub.status.busy": "2022-10-08T18:09:21.186980Z",
     "iopub.status.idle": "2022-10-08T18:09:21.954282Z",
     "shell.execute_reply": "2022-10-08T18:09:21.953324Z",
     "shell.execute_reply.started": "2022-10-08T18:09:21.187199Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "d56a45d2a2ce4a7db2bd7a4ebd301bba",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+-----------+-------------------+\n",
      "| id|       name|create_date|   last_update_time|\n",
      "+---+-----------+-----------+-------------------+\n",
      "|  1|Christopher| 2020-01-01|2020-01-02 00:00:00|\n",
      "|  2|       Will| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  3|   Emmeline| 2020-01-01|2020-01-02 00:00:00|\n",
      "|  4|       John| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  6|       Adam| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  7|       Bill| 2020-01-02|2020-01-03 00:00:00|\n",
      "+---+-----------+-----------+-------------------+"
     ]
    }
   ],
   "source": [
    "# Query current table as a point of comparison\n",
    "spark.sql(\" SELECT * FROM dev.db.iceberg_table \").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:09:24.232204Z",
     "iopub.status.busy": "2022-10-08T18:09:24.231959Z",
     "iopub.status.idle": "2022-10-08T18:09:25.006749Z",
     "shell.execute_reply": "2022-10-08T18:09:25.005847Z",
     "shell.execute_reply.started": "2022-10-08T18:09:24.232175Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "d05b16ac138a4f8f8c366d7b3fba3f50",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+-----------+-------------------+\n",
      "| id| name|create_date|   last_update_time|\n",
      "+---+-----+-----------+-------------------+\n",
      "|  1|Chris| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  2| Will| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  3| Emma| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  4| John| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  5| Eric| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  6| Adam| 2020-01-01|2020-01-01 00:00:00|\n",
      "+---+-----+-----------+-------------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\" SELECT * FROM dev.db.iceberg_table TIMESTAMP AS OF '2022-10-08 18:02:00' \").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-10-08T18:09:25.062935Z",
     "iopub.status.busy": "2022-10-08T18:09:25.062700Z",
     "iopub.status.idle": "2022-10-08T18:09:25.843329Z",
     "shell.execute_reply": "2022-10-08T18:09:25.842369Z",
     "shell.execute_reply.started": "2022-10-08T18:09:25.062911Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f4906f05e1664b0d8de74d83fb276a24",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+-----------+-------------------+\n",
      "| id| name|create_date|   last_update_time|\n",
      "+---+-----+-----------+-------------------+\n",
      "|  1|Chris| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  2| Will| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  3| Emma| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  4| John| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  5| Eric| 2020-01-01|2020-01-01 00:00:00|\n",
      "|  6| Adam| 2020-01-01|2020-01-01 00:00:00|\n",
      "+---+-----+-----------+-------------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\" SELECT * FROM dev.db.iceberg_table VERSION AS OF 4202173763955982227 \").show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "python",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}