{
"cells": [
{
"cell_type": "markdown",
"id": "b0ef68ac-9b8b-41d0-9e21-2f398bc6ea13",
"metadata": {
"execution": {
"iopub.execute_input": "2022-04-22T22:24:46.159901Z",
"iopub.status.busy": "2022-04-22T22:24:46.159679Z",
"iopub.status.idle": "2022-04-22T22:24:46.177881Z",
"shell.execute_reply": "2022-04-22T22:24:46.177080Z",
"shell.execute_reply.started": "2022-04-22T22:24:46.159876Z"
},
"tags": []
},
"source": [
"# Iceberg Example Using Hive Catalog Notebook"
]
},
{
"cell_type": "markdown",
"id": "e92709c0-3dfb-44f9-a4e7-e9f643fa98ff",
"metadata": {
"execution": {
"iopub.execute_input": "2022-04-22T22:24:46.159901Z",
"iopub.status.busy": "2022-04-22T22:24:46.159679Z",
"iopub.status.idle": "2022-04-22T22:24:46.177881Z",
"shell.execute_reply": "2022-04-22T22:24:46.177080Z",
"shell.execute_reply.started": "2022-04-22T22:24:46.159876Z"
},
"tags": []
},
"source": [
"## Topics covered in this example\n",
"\n",
"1) [Configuring Iceberg](#configure_iceberg)
\n",
"2) [Creating an Iceberg Table](#create_table)
\n",
"3) [DML Statements](#dml)
\n",
" a) [Inserts](#inserts)
\n",
" b) [Deletes](#deletes)
\n",
" d) [Updates](#updates)
\n",
"4) [Schema Evolution](#schema_evolution)
\n",
" a) [Adding Columns](#adding_columns)
\n",
" c) [Dropping Columns](#dropping_columns)
\n",
"5) [Time Travel](#time_travel)
\n",
" a) [Rollback](#rollback)
"
]
},
{
"cell_type": "markdown",
"id": "f0bac8b2-99cc-40e8-bce9-ac52e0c4a135",
"metadata": {
"execution": {
"iopub.execute_input": "2022-04-22T22:24:46.159901Z",
"iopub.status.busy": "2022-04-22T22:24:46.159679Z",
"iopub.status.idle": "2022-04-22T22:24:46.177881Z",
"shell.execute_reply": "2022-04-22T22:24:46.177080Z",
"shell.execute_reply.started": "2022-04-22T22:24:46.159876Z"
},
"tags": []
},
"source": [
"***\n",
"\n",
"## Prerequisites\n",
"\n",
"NOTE : In order to execute this notebook successfully as is, please ensure the following prerequisites are completed.
\n",
"\n",
"* This notebook was tested using a single node r5.4xlarge EMR 6.5 cluster. Iceberg 0.12.0, Spark 3.1.2\n",
"* To run this notebook\n",
" - Launch an EMR 6.5+ cluster in one of the subnets on which this EMR Studio is running.\n",
" - Launch the cluster with the following configuration classifications:\n",
"\n",
" \n",
" [\n",
" {\n",
" \"Classification\": \"iceberg-defaults \",\n",
" \"Properties\": {\n",
" \"iceberg.enabled\":\"true\"\n",
" }\n",
" }\n",
" ]\n",
"\n",
"The classification enables Iceberg.\n",
"\n",
"* Here is a sample CLI command to create an EMR cluster. Do remember to replace EMR-STUDIO-SUBNET with one of the subnets in which your EMR Studio is running:\n",
"\n",
" aws emr create-cluster \\\n",
" --name iceberg-emr-cluster\\\n",
" --use-default-roles \\\n",
" --release-label emr-6.5.0 \\\n",
" --instance-count 1 \\\n",
" --instance-type r5.4xlarge \\\n",
" --applications Name=Hadoop Name=Livy Name=Spark Name=JupyterEnterpriseGateway \\\n",
" --ec2-attributes SubnetId=\\\n",
" --configurations '[{\"Classification\":\"iceberg-defaults\",\"Properties\":{\"iceberg.enabled\":\"true\"}}]'\n",
" \n",
"\n",
"* This notebook uses the `PySpark` kernel. \n",
"***"
]
},
{
"cell_type": "markdown",
"id": "04e47cf4-bebd-4425-9aed-44ddc8a35d76",
"metadata": {
"tags": []
},
"source": [
"## Introduction\n",
"Apache Iceberg (https://iceberg.apache.org/) is an open table format for huge analytic datasets. Iceberg adds tables to compute engines including Spark, Trino, PrestoDB, Flink and Hive using a high-performance table format that works just like a SQL table. Iceberg tracks individual data files in a table instead of directories. This allows writers to create data files in-place and only adds files to the table in an explicit commit. Every time a new file is inserted to any partition in this table, a new point-in-time snapshot of all the files get created. At the query time, there is no need to list a directory to find the files we need to work with, as the snapshot already has that information pre-populated during the write time (commonly known as snapshot isolation (https://en.wikipedia.org/wiki/Snapshot_isolation) in databases).\n",
"\n",
"Iceberg supports write, delete, update, and time travel operations with complete support for ACID transactions (https://en.wikipedia.org/wiki/ACID). Table changes are atomic and readers never see partial or uncommitted changes (serializable isolation (https://en.wikipedia.org/wiki/Isolation_(database_systems)#Serializable))\n",
"\n",
"Iceberg table format is an open specification at multiple levels. At the catalog level, you can plugin multiple types of catalogs such as hive, hadoop, AWS Glue Data Catalog etc. All these can co-exist. You can join tables across different types of catalogs. In this example, we are going to work with Hive Data Catalog.\n",
"\n",
"***"
]
},
{
"cell_type": "markdown",
"id": "deca2149-344c-4baf-a3b6-4bd19d7afbcb",
"metadata": {
"tags": []
},
"source": [
"## Setup\n",
"Create an S3 bucket location to save sample dataset. In this example we use the path format: s3://my_s3_bucket/iceberg/YOUR-CATALOG-NAME/tables/ \n",
" \n",
" For example: s3://my_s3_bucket/iceberg/db/amazon_reviews_iceberg\n",
"\n",
"update the my_s3_bucket with the bucket which you create.\n",
"***"
]
},
{
"cell_type": "markdown",
"id": "5af5f2a5-bd04-461b-8617-1f9fcfe92e83",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-09T23:20:08.528217Z",
"iopub.status.busy": "2022-05-09T23:20:08.527988Z"
},
"tags": []
},
"source": [
"\n",
"\n",
"\n",
"## Configuring Iceberg on Spark session\n",
"\n",
"Configure your Spark session using the %%configure magic command. We will be using Hive Catalog for Iceberg Tables\n",
"\n",
"Update the my_bucket_name in below configuration which you create the Prerequisite - Create S3 Bucket section of workshop"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "b42dd813-a4a5-42da-b29c-bbf9e02d9046",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"%%configure -f\n",
"{\n",
"\"conf\":{\n",
" \"spark.sql.extensions\":\"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions\",\n",
" \"spark.sql.catalog.spark_catalog\":\"org.apache.iceberg.spark.SparkSessionCatalog\",\n",
" \"spark.sql.catalog.spark_catalog.type\":\"hive\",\n",
" \"spark.sql.catalog.dev\":\"org.apache.iceberg.spark.SparkCatalog\",\n",
" \"spark.sql.catalog.dev.type\":\"hadoop\",\n",
" \"spark.sql.catalog.dev.warehouse\":\"s3://my_s3_bucket/iceberg/\"\n",
" }\n",
"}"
]
},
{
"cell_type": "markdown",
"id": "8725d217-b26c-4fa8-84de-18969f5260c7",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-09T23:32:45.009182Z",
"iopub.status.busy": "2022-05-09T23:32:45.008967Z",
"iopub.status.idle": "2022-05-09T23:32:45.078427Z",
"shell.execute_reply": "2022-05-09T23:32:45.077766Z",
"shell.execute_reply.started": "2022-05-09T23:32:45.009159Z"
},
"tags": []
},
"source": [
"\n",
"## Create Iceberg Table, this table is using Hive Catalog. \n",
"\n",
"Update the **my_bucket_name** in below DDL.\n",
"\n",
"We will use PySpark for most of our Iceberg operations, although you could use equivalent Spark SQL, Scala, or Java languages to achieve all of these as well."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "23defea5-893c-4681-89cd-e35f85ff4259",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\" DROP TABLE if exists dev.db.amazon_reviews_iceberg\"\"\")\n",
"\n",
"spark.sql(\"\"\" CREATE TABLE dev.db.amazon_reviews_iceberg (\n",
" marketplace string,\n",
" customer_id string,\n",
" review_id string,\n",
" product_id string,\n",
" product_parent string,\n",
" product_title string,\n",
" star_rating int,\n",
" helpful_votes int,\n",
" total_votes int,\n",
" vine string,\n",
" verified_purchase string,\n",
" review_headline string,\n",
" review_body string,\n",
" review_date date,\n",
" year int)\n",
"USING iceberg \n",
"location 's3://my_s3_bucket/iceberg/db/amazon_reviews_iceberg'\"\"\")"
]
},
{
"cell_type": "markdown",
"id": "a91913e0-a89c-464d-86d8-713929dcb2c1",
"metadata": {
"execution": {
"iopub.execute_input": "2022-04-22T22:26:08.320405Z",
"iopub.status.busy": "2022-04-22T22:26:08.320101Z",
"iopub.status.idle": "2022-04-22T22:26:15.660833Z",
"shell.execute_reply": "2022-04-22T22:26:15.660248Z",
"shell.execute_reply.started": "2022-04-22T22:26:08.320372Z"
},
"tags": []
},
"source": [
"\n",
"## DML Operations\n",
"Icerberg supports all DML statements to add or modify data in your data lake: Inserts to add new data, Updates to modify specific columns in specific rows in your existing data, Deletes for GDPR and CCPA compliance and Upserts when you have incoming data that may have a mix of inserts and updates. Let us look at each of them now."
]
},
{
"cell_type": "markdown",
"id": "58730a9c-431d-49b8-985f-ef2177f439a2",
"metadata": {
"execution": {
"iopub.execute_input": "2022-04-22T22:26:08.320405Z",
"iopub.status.busy": "2022-04-22T22:26:08.320101Z",
"iopub.status.idle": "2022-04-22T22:26:15.660833Z",
"shell.execute_reply": "2022-04-22T22:26:15.660248Z",
"shell.execute_reply.started": "2022-04-22T22:26:08.320372Z"
},
"tags": []
},
"source": [
"\n",
"### Inserts"
]
},
{
"cell_type": "markdown",
"id": "fffdbc34-983d-44ad-b606-0f5d7d434158",
"metadata": {},
"source": [
"**We will be using Amazon Product Reviews Dataset dataset, spend some time to get familiarized with this dataset.**\n",
"\n",
"We are loading just one partition for sake of simplicity"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "7c898a47-33eb-4882-9990-e2dbaf257d68",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"df = spark.read.parquet(\"s3://amazon-reviews-pds/parquet/product_category=Electronics/*.parquet\")"
]
},
{
"cell_type": "markdown",
"id": "bb85126b-117e-4ddd-b0b6-8652e6025e95",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-09T23:30:25.250417Z",
"iopub.status.busy": "2022-05-09T23:30:25.250053Z",
"iopub.status.idle": "2022-05-09T23:30:25.323478Z",
"shell.execute_reply": "2022-05-09T23:30:25.320426Z",
"shell.execute_reply.started": "2022-05-09T23:30:25.250393Z"
},
"tags": []
},
"source": [
"**Run below cell to write data into the Iceberg table, We are writing just one partition for sake of simplicity**"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4d1ac4ee-7b08-4842-9d53-99bb5b22cfa6",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"df.writeTo(\"dev.db.amazon_reviews_iceberg\").append()"
]
},
{
"cell_type": "markdown",
"id": "87703835-305a-46e5-89d1-26a81645df83",
"metadata": {
"tags": []
},
"source": [
"**Verify data is loaded into iceberg table successfully.**"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e211cc44-ced2-4c50-b957-f01517777e24",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"select * from dev.db.amazon_reviews_iceberg\"\"\").show()"
]
},
{
"cell_type": "markdown",
"id": "3190ca8c-803c-42f4-98b9-62c26be29959",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-09T23:51:57.948259Z",
"iopub.status.busy": "2022-05-09T23:51:57.948019Z",
"iopub.status.idle": "2022-05-09T23:51:58.018887Z",
"shell.execute_reply": "2022-05-09T23:51:58.018177Z",
"shell.execute_reply.started": "2022-05-09T23:51:57.948235Z"
}
},
"source": [
"\n",
"### Deletes\n",
"GDPR and CCPA regulations mandate timely removal of individual customer data and other records from datasets. Iceberg is designed to be able to handle these trivially.\n",
"Now let us delete a record from our Iceberg table."
]
},
{
"cell_type": "markdown",
"id": "097e2138-1059-420e-a916-f4a21dd159dc",
"metadata": {
"tags": []
},
"source": [
"Delete all records from the table for verified_purchase = 'N'"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4054056c-fae1-44b9-ba89-3243ce0d8ea5",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"delete from dev.db.amazon_reviews_iceberg\n",
"where verified_purchase = 'N'\"\"\")"
]
},
{
"cell_type": "markdown",
"id": "5d8d0950-c909-4177-bd80-c725f577e996",
"metadata": {
"tags": []
},
"source": [
"Test if data is deleted. Below query should produce zero records."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "653dc08f-b2be-4bec-afce-4b9a88f19712",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"select * from dev.db.amazon_reviews_iceberg where verified_purchase = 'N'\"\"\").show()"
]
},
{
"cell_type": "markdown",
"id": "4504fd86-29c5-4c5f-a71d-f7e9643db3c8",
"metadata": {},
"source": [
"\n",
"### Updates\n",
"What if we want to go back and update an existing record? Let's change the `marketplace` from US to USA. Iceberg allows updates using a simple `UPDATE` and`SET` clause added to your query"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "cdf9839a-a352-4aca-9de7-59a784d4373a",
"metadata": {},
"outputs": [],
"source": [
"spark.sql(\"\"\"UPDATE dev.db.amazon_reviews_iceberg\n",
"SET marketplace = 'USA'\n",
"WHERE marketplace = 'US'\"\"\")"
]
},
{
"cell_type": "markdown",
"id": "d973db13-d010-485d-82a9-cc2c81b3fb13",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-10T00:32:39.531722Z",
"iopub.status.busy": "2022-05-10T00:32:39.531500Z",
"iopub.status.idle": "2022-05-10T00:32:41.829618Z",
"shell.execute_reply": "2022-05-10T00:32:41.828751Z",
"shell.execute_reply.started": "2022-05-10T00:32:39.531700Z"
}
},
"source": [
"Verify 'marketplace' column is updated"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "19bdb3bd-b7d3-458c-8fff-2e0dbc035eb7",
"metadata": {},
"outputs": [],
"source": [
"spark.sql(\"\"\"select * from dev.db.amazon_reviews_iceberg\"\"\").show()"
]
},
{
"cell_type": "markdown",
"id": "85f8e445-d0a1-4bb1-9dfe-f1bf0ff946e1",
"metadata": {},
"source": [
"\n",
"## Schema Evolution\n",
"Borrowing from the way columns work in databases, Iceberg tracks columns by using unique IDs and not by the column name. As long as the ID is the same, all the data still remains. You can safely add, drop, rename, update, or even reorder columns. You don’t have to rewrite the data for this. Schema evolution gets first class citizen treatment in Iceberg. Your ingest and read queries now have the freedom to be evolved without having to hide the schema inside JSON blobs."
]
},
{
"cell_type": "markdown",
"id": "c8c2d994-ea82-421e-aeb5-9894bcd8fd68",
"metadata": {
"tags": []
},
"source": [
"In this example we will add a column to the iceberg table which we just created. We will add comment column to the table."
]
},
{
"cell_type": "markdown",
"id": "26b3f398-af14-4002-9ab5-5f084e81c361",
"metadata": {
"tags": []
},
"source": [
"\n",
"### Adding Columns\n",
"Now we are going to add another column called `high_rated_product`. Iceberg also allows documenting the purpose for each column as `comment`, which helps a lot in a collaborative environment and quick lookup of data from business users."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9b34f116-db73-4f80-92f4-10400a717066",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"ALTER TABLE dev.db.amazon_reviews_iceberg ADD COLUMNS (high_rated_product string comment 'Highly rated comment')\"\"\")"
]
},
{
"cell_type": "markdown",
"id": "81883f1c-47af-4dfa-9a62-cbf647c03d92",
"metadata": {
"tags": []
},
"source": [
"We will add **High rated** flag to the comment column where rating is greater or equal to 4"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "fb7863b4-a453-490d-a638-7cb29368a76f",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"UPDATE dev.db.amazon_reviews_iceberg SET high_rated_product = 'High rated'Where star_rating >=4\"\"\")"
]
},
{
"cell_type": "markdown",
"id": "34824f0a-4334-45d3-8a0d-9a575875488a",
"metadata": {
"tags": []
},
"source": [
"Verify column is added successfully by quering the table."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "91cd1eac-c422-4c44-bdf7-d23284bc87d6",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"Select customer_id,review_id,product_id, product_title, star_rating, high_rated_product from dev.db.amazon_reviews_iceberg\"\"\").show()"
]
},
{
"cell_type": "markdown",
"id": "9d19f6c3-9f08-402f-85ac-8aa487f51c05",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-10T04:46:55.786156Z",
"iopub.status.busy": "2022-05-10T04:46:55.785922Z",
"iopub.status.idle": "2022-05-10T04:46:57.073122Z",
"shell.execute_reply": "2022-05-10T04:46:57.072441Z",
"shell.execute_reply.started": "2022-05-10T04:46:55.786133Z"
},
"tags": []
},
"source": [
"\n",
"### Dropping Columns\n",
"Now, there is a change in business requirements, we are not interested in the `high_rated_product` column anymore and need to remove that column from our table. Iceberg allows us to do that easily."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d338d82a-f7b0-4ef7-8e1b-ebb2c88c9081",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"ALTER TABLE dev.db.amazon_reviews_iceberg DROP COLUMN high_rated_product\"\"\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "906629e3-5c0d-4ab6-a2dc-5d8a205c2288",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"\"\"select * from dev.db.amazon_reviews_iceberg\"\"\").show()"
]
},
{
"cell_type": "markdown",
"id": "fb22bf8b-3c83-45b1-aac6-c37e61b41e20",
"metadata": {
"execution": {
"iopub.execute_input": "2022-04-25T22:32:33.100830Z",
"iopub.status.busy": "2022-04-25T22:32:33.100599Z",
"iopub.status.idle": "2022-04-25T22:32:33.160097Z",
"shell.execute_reply": "2022-04-25T22:32:33.159539Z",
"shell.execute_reply.started": "2022-04-25T22:32:33.100805Z"
},
"tags": []
},
"source": [
"\n",
"## Time Travel\n",
"Let us query our table as of the previous snapshot."
]
},
{
"cell_type": "markdown",
"id": "24eb246b-b759-49de-9364-6831ab567ec4",
"metadata": {
"execution": {
"iopub.execute_input": "2022-04-25T21:15:16.899280Z",
"iopub.status.busy": "2022-04-25T21:15:16.899007Z",
"iopub.status.idle": "2022-04-25T21:15:17.189559Z",
"shell.execute_reply": "2022-04-25T21:15:17.188833Z",
"shell.execute_reply.started": "2022-04-25T21:15:16.899247Z"
},
"tags": []
},
"source": [
"**To View Table's History, Snapshot, Mainefest and Metadata**"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "b129db58-0002-49b0-99b9-71f6c2a458e8",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"#Table History\n",
"spark.sql(\"\"\"SELECT * FROM dev.db.amazon_reviews_iceberg.history\"\"\").show()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "72fbe91b-d985-4313-8239-e84e4355dded",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"#Table Snapshot\n",
"spark.sql(\"\"\"SELECT * FROM dev.db.amazon_reviews_iceberg.snapshots\"\"\").show()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e3bb55b4-0aea-4f7c-a2f8-055ab101852e",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"#You can also join snapshots to table history. For example, this query will show table history, with the application ID that wrote each snapshot:\n",
"spark.sql(\"\"\"select\n",
" h.made_current_at,\n",
" s.operation,\n",
" h.snapshot_id,\n",
" h.is_current_ancestor,\n",
" s.summary['spark.app.id']\n",
"from dev.db.amazon_reviews_iceberg.history h\n",
"join dev.db.amazon_reviews_iceberg.snapshots s\n",
" on h.snapshot_id = s.snapshot_id\n",
"order by made_current_at\"\"\").show()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c8bd9869-3549-44e3-ac9e-ae8568a9cb87",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"#To show a table’s data files and each file’s metadata, run:\n",
"spark.sql(\"\"\"SELECT * FROM dev.db.amazon_reviews_iceberg.files\"\"\").show()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ec8efc76-f549-4de3-a06d-ebe5e04e2fdd",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"#To show a table’s file manifests and each file’s metadata, run:\n",
"spark.sql(\"\"\"SELECT * FROM dev.db.amazon_reviews_iceberg.manifests\"\"\").show()"
]
},
{
"cell_type": "markdown",
"id": "1d035fa4-6fc5-4c1c-b846-3c386ca0222b",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-10T05:10:28.203773Z",
"iopub.status.busy": "2022-05-10T05:10:28.203513Z",
"iopub.status.idle": "2022-05-10T05:10:28.254487Z",
"shell.execute_reply": "2022-05-10T05:10:28.253935Z",
"shell.execute_reply.started": "2022-05-10T05:10:28.203748Z"
},
"tags": []
},
"source": [
"\n",
"### Rollback\n",
"\n",
"To undo the recent changes, we can execute Iceberg stored procedures using `CALL` statement to rollback the state of the table to any historical commit using `rollback_to_snapshot` stored procedure. We could also use `rollback_to_timestamp`.\n",
"\n",
"Recover the table to its original state, replace the xxxxxxxxxxxxx with Snapshot id from Table History. Use the snapshot_id with parent_id = null from Table History (first record)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "3073f8a6-121f-4768-b112-7ebae68d28f4",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"CALL dev.system.rollback_to_snapshot('db.amazon_reviews_iceberg', xxxxxxxxxxxxx)\")"
]
},
{
"cell_type": "markdown",
"id": "d4f8bc24-5ad0-4b1b-8e94-34d65b6d62c3",
"metadata": {
"execution": {
"iopub.execute_input": "2022-05-10T18:47:19.451231Z",
"iopub.status.busy": "2022-05-10T18:47:19.451013Z",
"iopub.status.idle": "2022-05-10T18:47:28.739582Z",
"shell.execute_reply": "2022-05-10T18:47:28.738989Z",
"shell.execute_reply.started": "2022-05-10T18:47:19.451207Z"
},
"tags": []
},
"source": [
"Our table is now back to original state"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "3dce569e-93c4-43c3-a962-c72fcc83f8ef",
"metadata": {
"tags": []
},
"outputs": [],
"source": [
"spark.sql(\"select * from dev.db.amazon_reviews_iceberg\").show()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "PySpark",
"language": "",
"name": "pysparkkernel"
},
"language_info": {
"codemirror_mode": {
"name": "python",
"version": 3
},
"mimetype": "text/x-python",
"name": "pyspark",
"pygments_lexer": "python3"
}
},
"nbformat": 4,
"nbformat_minor": 5
}