01_standalone_examples/azure-databricks/Delta Lake Demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "57de49db-6f15-4676-8df3-471926743528",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## [Integration of lakeFS with Delta Lake](https://docs.lakefs.io/integrations/delta.html)\n",
    "\n",
    "### Use Cases:\n",
    "#### 1. Isolating ETL job and atomic promotion to production\n",
    "#### 2. Atomic rollback of Multi-Table Transactions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6ed0d6d1-723c-438e-b2fc-a5b132449d3e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Prerequisites\n",
    "\n",
    "####### This Notebook requires connecting to a lakeFS Server. \n",
    "####### To spin up lakeFS quickly - use [lakeFS Cloud](https://demo.lakefs.io) which provides lakeFS server on-demand with a single click; \n",
    "####### Or, alternatively, refer to [lakeFS Quickstart doc](https://docs.lakefs.io/quickstart/installing.html)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1d0d2b56-5564-477f-ac5b-d72670975a25",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Setup Task: Change your lakeFS credentials"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e1cf7b5a-5852-415b-9fe1-7e156199045e",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "lakefsEndPoint = '<lakeFS Endpoint URL>'\n",
    "lakefsAccessKey = '<lakeFS Access Key>'\n",
    "lakefsSecretKey = '<lakeFS Secret Key>'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5d037c6c-58f6-43cb-b77d-251b50441c07",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Setup Task: You can change lakeFS repo name (it can be an existing repo or provide a new repo name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "717cef39-e839-4c33-8a5e-a21a02d0fe69",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "repositoryName = \"delta-repo\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3202c26a-7f9d-4d46-855c-89e1d57fe84a",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Setup Task: Versioning Information"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6d7f0d59-4249-4425-a702-ebd35e3de9cb",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "mainBranch = \"main\"\n",
    "deltaLakeETLBranch = \"delta-lake-etl-branch\"\n",
    "customersTable = \"customers\"\n",
    "ordersTable = \"orders\"\n",
    "orderUpdatesTable = \"order_updates\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b5d816ba-cb78-41db-9eab-20bafc8d2758",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Setup Task: Storage Information - Optional on Playground\n",
    "###### Change the Storage Namespace to a location in the bucket you’ve configured. The storage namespace is a location in the underlying storage where data for this repository will be stored."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "602c6a25-df9e-4e14-865d-2368dc59352c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "storageNamespace = \"https://storage-account-name.blob.core.windows.net/storage-container-name/\"  + repositoryName # Unique per repository"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "4bb0c548-53da-457c-a44f-d388e1f5d0fb",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Setup Task: This notebook uses few sample CSV data files. Download all CSV files from [lakeFS samples data set](https://github.com/treeverse/lakeFS-samples/tree/main/data/OrionStar) and upload to [Databricks Filestore](https://docs.databricks.com/dbfs/filestore.html) in '/FileStore/tables/data/OrionStar' folder. If you use a different folder name then change the folder name here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e9274953-3aeb-4578-956d-9ab2772bb470",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "sampleDataFolder = '/FileStore/tables/data/OrionStar'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "b1a2f7a0-b356-4329-9569-18317071cc41",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Setup Task: Run additional [Setup](./?o=8911673420610391#notebook/3578552658771058) tasks here"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a73a54ef-40a7-4dbe-ac0b-50a277f5066b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%run ./deltaLakeSetup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "016bf741-9293-4224-a595-308c5d747f65",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Create Repository - Optional on Playground or if repository exists"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a64f9dfb-414b-4b21-8b4b-67a64a38cee2",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "repo = lakefs.Repository(\n",
    "  repositoryName,\n",
    "  client=clt).create(\n",
    "    storage_namespace=storageNamespace,\n",
    "    default_branch=mainBranch,\n",
    "    exist_ok=True)\n",
    "branchMain = repo.branch(mainBranch)\n",
    "print(repo)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "74fc977e-ba8e-4b92-891b-2972f91db025",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Create Customers delta table in the main branch (using CUSTOMER.csv file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b07d8005-8bda-4eca-947a-97c4b955e58b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{customersTable}\"\n",
    "df = spark.read.csv(sampleDataFolder+'/CUSTOMER.csv',header=True,schema=customersSchema)\n",
    "df.write.format(\"delta\").mode(\"overwrite\").save(customersTablePath)\n",
    "df.display(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "321a032d-1ffb-4228-89cd-9b8c6f4a5640",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Create Orders delta table in the main branch (using ORDER_FACT.csv file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d6bc3160-dffc-4f75-b378-6a0ae4903e2f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.csv(sampleDataFolder+'/ORDER_FACT.csv',header=True,schema=ordersSchema)\n",
    "df.write.format(\"delta\").mode(\"overwrite\").save(ordersTablePath)\n",
    "df.display(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "9c43cd54-1dd1-4000-92fe-c6217f4ef311",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Commit changes and attach some metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b598eea4-6220-446f-aa8d-43da9abe6b74",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "branchMain.commit(message='Added customers and orders Delta tables!', \n",
    "        metadata={'using': 'python_api'})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "6f65d05b-2344-407d-a893-ef65be74ce99",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### ETL Job Starts"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "a771986c-fffb-435c-9ab1-b32133595761",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Create a new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "87bc5832-e29e-4227-a6c2-20d81ca2a291",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "branchDeltaLakeETL = repo.branch(deltaLakeETLBranch).create(source_reference=mainBranch, exist_ok=True)\n",
    "print(f\"{deltaLakeETLBranch} ref:\", branchDeltaLakeETL.get_commit().id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "9e46159f-72e8-42be-acf1-98ea93da7912",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Apply POS (Point of Sale) Transactions to Delta Lake: delete data for a customer on the new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "da216456-1244-48a5-9e8b-717c1fc7f17c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from delta.tables import *\n",
    "\n",
    "ordersTablePath = f\"lakefs://{repositoryName}/{deltaLakeETLBranch}/{ordersTable}\"\n",
    "deltaTable = DeltaTable.forPath(spark, ordersTablePath)\n",
    "deltaTable.delete(\"Customer_ID = 19444\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "208d94b0-b5a7-4795-9d33-5c6c62ea4298",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"lakefs://{repositoryName}/{deltaLakeETLBranch}/{customersTable}\"\n",
    "deltaTable = DeltaTable.forPath(spark, customersTablePath)\n",
    "deltaTable.delete(\"Customer_ID = 19444\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "5b29e468-28ec-4823-a4ab-b286535b346a",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Apply POS Transactions to Delta Lake: update data for a customer on the new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2fa9acc2-dbd0-4bc4-9626-c0aa1f37b2c4",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"lakefs://{repositoryName}/{deltaLakeETLBranch}/{customersTable}\"\n",
    "deltaTable = DeltaTable.forPath(spark, customersTablePath)\n",
    "deltaTable.update(\n",
    "  condition = expr(\"Customer_ID = 63\"),\n",
    "  set = { \"Customer_FirstName\": \"'Jim'\",\n",
    "          \"Customer_Name\": \"'Jim Klisurich'\"})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "ca480adb-3fef-4232-b8ee-9f7dadbcb1e6",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Apply POS Transactions to Delta Lake: batch upsert (5 updated and 10 new orders in ORDER_FACT_UPDATES.csv file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a36f4e1a-151b-4615-9089-77b9c2c0263f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"lakefs://{repositoryName}/{deltaLakeETLBranch}/{ordersTable}\"\n",
    "deltaTableOrders = DeltaTable.forPath(spark, ordersTablePath)\n",
    "\n",
    "orderUpdatesTablePath = f\"lakefs://{repositoryName}/{deltaLakeETLBranch}/{orderUpdatesTable}\"\n",
    "dfOrderUpdates = spark.read.csv(sampleDataFolder+'/ORDER_FACT_UPDATES.csv',header=True,schema=ordersSchema)\n",
    "dfOrderUpdates.write.format(\"delta\").mode(\"overwrite\").save(orderUpdatesTablePath)\n",
    "\n",
    "deltaTableOrders.alias('orders') \\\n",
    "  .merge(\n",
    "    dfOrderUpdates.alias('orderUpdates'),\n",
    "    'orders.Order_ID = orderUpdates.Order_ID AND orders.Product_ID = orderUpdates.Product_ID'\n",
    "  ) \\\n",
    "  .whenMatchedUpdate(set =\n",
    "    {\n",
    "      \"Customer_ID\": \"orderUpdates.Customer_ID\",\n",
    "      \"Employee_ID\": \"orderUpdates.Employee_ID\",\n",
    "      \"Street_ID\": \"orderUpdates.Street_ID\",\n",
    "      \"Order_Date\": \"orderUpdates.Order_Date\",\n",
    "      \"Delivery_Date\": \"orderUpdates.Delivery_Date\",\n",
    "      \"Order_ID\": \"orderUpdates.Order_ID\",\n",
    "      \"Order_Type\": \"orderUpdates.Order_Type\",\n",
    "      \"Product_ID\": \"orderUpdates.Product_ID\",\n",
    "      \"Quantity\": \"orderUpdates.Quantity\",\n",
    "      \"Total_Retail_Price\": \"orderUpdates.Total_Retail_Price\",\n",
    "      \"CostPrice_Per_Unit\": \"orderUpdates.CostPrice_Per_Unit\",\n",
    "      \"Discount\": \"orderUpdates.Discount\"\n",
    "    }\n",
    "  ) \\\n",
    "  .whenNotMatchedInsert(values =\n",
    "    {\n",
    "      \"Customer_ID\": \"orderUpdates.Customer_ID\",\n",
    "      \"Employee_ID\": \"orderUpdates.Employee_ID\",\n",
    "      \"Street_ID\": \"orderUpdates.Street_ID\",\n",
    "      \"Order_Date\": \"orderUpdates.Order_Date\",\n",
    "      \"Delivery_Date\": \"orderUpdates.Delivery_Date\",\n",
    "      \"Order_ID\": \"orderUpdates.Order_ID\",\n",
    "      \"Order_Type\": \"orderUpdates.Order_Type\",\n",
    "      \"Product_ID\": \"orderUpdates.Product_ID\",\n",
    "      \"Quantity\": \"orderUpdates.Quantity\",\n",
    "      \"Total_Retail_Price\": \"orderUpdates.Total_Retail_Price\",\n",
    "      \"CostPrice_Per_Unit\": \"orderUpdates.CostPrice_Per_Unit\",\n",
    "      \"Discount\": \"orderUpdates.Discount\"\n",
    "    }\n",
    "  ) \\\n",
    "  .execute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "c9ebcd0b-19b4-419e-8d17-e7695aed0c0e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Customers delta table in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b2516d10-8d1c-4c46-ad72-c5e7ce8a8f77",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{customersTable}\"\n",
    "df = spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").display()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1fc208e7-ddb7-459c-b5bf-077698e1dd6b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"lakefs://{repositoryName}/{deltaLakeETLBranch}/{customersTable}\"\n",
    "spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").display()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "b337d871-bc52-4e33-8db9-afc7b2c25205",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Customers count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f00ff693-22e7-4c45-9c4d-2a1f125a3f43",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(customersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "cac554b2-84ed-4513-8329-69bcee6176dc",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Orders delta table in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2cc3147c-a837-4776-a77b-9ade0423af35",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").display()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b874f7f8-6fdd-4fd4-a8b5-04e9df823e2c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"lakefs://{repositoryName}/{deltaLakeETLBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").display()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "23cdbf8c-47ef-46f1-ae9d-69e00f4eedbb",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Orders count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "025df9fc-64bf-44ae-813d-07d0abc82928",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(ordersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "a4fabaf8-ee29-4b90-893d-e7f7aeb2e628",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Commit changes and attach some metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5e436c6d-921e-4cd8-bbfd-21cd6a1e355d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "branchDeltaLakeETL.commit(message='Deleted and updated customers. Deleted and upserted orders.', \n",
    "        metadata={'using': 'python_api'})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "689ef3d5-b9ab-4614-88b4-de17bc3e6ca5",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Diff between the new branch and the source branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c98b2301-4e94-420e-bad0-26fa24cc5bdf",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "diff = branchMain.diff(other_ref=branchDeltaLakeETL)\n",
    "print_diff(diff)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "63ad945d-1257-4bd8-98a6-b1586d62d15f",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### ETL Job Completes\n",
    "\n",
    "#### Delete new branch if ETL job fails or merge new branch to main branch if ETL job succeeds\n",
    "\n",
    "#### Delete new branch if ETL job fails"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d135828a-0929-4291-8225-fce48d9e459d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Uncomment if you want to run this\n",
    "\n",
    "#branchDeltaLakeETL.delete()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "88a0385d-ffec-4261-8327-8fe3f333e9c8",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Or merge new branch to the main branch if ETL job succeeds (atomic promotion to production)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d7243079-fb6e-40f7-97d5-1979edf5c3b3",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "res = branchDeltaLakeETL.merge_into(branchMain)\n",
    "print(res)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "a9f302a2-1659-4dab-b9f4-4eb67bf33404",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Read data from the main branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "bd6e2711-d694-4feb-bf73-a07efb9d6c2f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{customersTable}\"\n",
    "spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").display()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f9b2d997-feb3-48f0-8b09-64abd6787c51",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").display()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "0cc868de-17bf-4e36-9bf6-887cdfba0b01",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Customers count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b70a7ce1-7717-4179-ba5d-273057b791fa",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(customersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "a852ccd7-0d43-45e1-babd-aa7f21cb3972",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Orders count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2304cebb-a0b7-4dcb-ac5f-0be8f0517be2",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(ordersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "1802a68c-83b4-489e-a2f4-77fcfd362b88",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### If you merged new branch to the main branch then you can atomically rollback Multi-Table Transactions\n",
    "\n",
    "##### Go to lakeFS UI and get the commit ID or copy the 'reference' from the previous merge statement"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "aeaef858-4ff7-41e9-84ee-4bc06bf0f0fd",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "branchMain.revert(parent_number=1, reference=mainBranch)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "c9118fa3-8091-4ac7-96a2-00be1a721985",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Read data again from the main branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "13ee9b28-43e8-44c6-b755-5d875e4d0213",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{customersTable}\"\n",
    "spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").display()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b5488af1-4d8f-4c85-bfb7-b0cef6998517",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"lakefs://{repositoryName}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").display()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "a6a16ebb-2778-429c-913b-8e6d4bfba458",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Customers count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b4d230be-47a3-44d4-b65d-219ccb336584",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(customersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "2e32dbd9-2706-41c6-840c-ea38bd74b5be",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### Data Validation: Compare Orders count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "125b6cf9-f6c5-447a-a34c-e7f117521a96",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(ordersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "787c4d80-a5a1-41f1-9a26-0d7fb8f51756",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### More Questions?\n",
    "\n",
    "###### Join the [lakeFS Slack group](https://lakefs.io/slack)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "251340f1-35fe-4cb4-bf76-d7beeab7e45b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "Delta Lake Demo",
   "widgets": {}
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}