00_notebooks/delta-lake.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c0941866-5ab1-46c6-ae4e-3ae228311623",
   "metadata": {},
   "source": [
    "<img src=\"./images/logo.svg\" alt=\"lakeFS logo\" width=300/> \n",
    "\n",
    "# Integration of lakeFS with Delta Lake\n",
    "\n",
    "[📚 Docs](https://docs.lakefs.io/integrations/delta.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "29105bda-b8b6-4bff-aaa8-7eadbacc7923",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Use Cases:\n",
    "\n",
    "1. Isolating ETL job and atomic promotion to production\n",
    "2. Atomic rollback of Multi-Table Transactions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "531dde95-2a5b-4625-becf-55aaf9c496ec",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Config\n",
    "\n",
    "**_If you're not using the provided lakeFS server and MinIO storage then change these values to match your environment_**"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "61c5aea5-5195-4caa-a535-95b9bab8a1be",
   "metadata": {},
   "source": [
    "### lakeFS endpoint and credentials"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "79a78d9a-10b0-4ffa-b501-213876abe1ad",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "lakefsEndPoint = 'http://lakefs:8000' # e.g. 'https://username.aws_region_name.lakefscloud.io' \n",
    "lakefsAccessKey = 'AKIAIOSFOLKFSSAMPLES'\n",
    "lakefsSecretKey = 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31ca7f50-bd3d-4a86-a4ed-30066b42c971",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Object Storage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f6a5b298-1879-4441-a607-0e64289bf90e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "storageNamespace = 's3://example' # e.g. \"s3://bucket\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "853dd30d-ec0e-4302-9dd2-97c4fa21257e",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7e8191ce-2e9a-4eda-a107-ae99659a5397",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Setup\n",
    "\n",
    "**(you shouldn't need to change anything in this section, just run it)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a45fbab3-5f98-46ad-8e12-9021ed2bf81a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "repo_name = \"delta-lake-demo\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bcf0d58d-8465-4e19-b572-f1bf48f61e2d",
   "metadata": {},
   "source": [
    "### Versioning Information"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98f11740-f385-42ef-b8e9-a3d89484b1ca",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "mainBranch = \"main\"\n",
    "deltaLakeETLBranch = \"delta-lake-etl-branch\"\n",
    "customersTable = \"customers\"\n",
    "ordersTable = \"orders\"\n",
    "orderUpdatesTable = \"order_updates\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85d3547a-7c4c-48b5-ab8b-871cb1c2a230",
   "metadata": {},
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "692dfb5c-f7de-45a7-b3dd-f4ceb1898bd5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import lakefs\n",
    "from assets.lakefs_demo import print_commit, print_diff\n",
    "from pyspark.sql.types import ByteType, IntegerType, LongType, StringType, StructType, StructField\n",
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eaa5bb3f-7432-4637-bc8e-68159bdcb969",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Set environment variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4795b1e7-761e-4f7a-840b-982f99ff3c6a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "os.environ[\"LAKECTL_SERVER_ENDPOINT_URL\"] = lakefsEndPoint\n",
    "os.environ[\"LAKECTL_CREDENTIALS_ACCESS_KEY_ID\"] = lakefsAccessKey\n",
    "os.environ[\"LAKECTL_CREDENTIALS_SECRET_ACCESS_KEY\"] = lakefsSecretKey"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "277bb5fa-37b5-4b53-b991-30e11af7e51a",
   "metadata": {},
   "source": [
    "#### Verify lakeFS credentials by getting lakeFS version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8f7005b1-2cb0-40fd-b099-5059bd00a8d9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "print(\"Verifying lakeFS credentials…\")\n",
    "try:\n",
    "    v=lakefs.client.Client().version\n",
    "except:\n",
    "    print(\"🛑 failed to get lakeFS version\")\n",
    "else:\n",
    "    print(f\"…✅lakeFS credentials verified\\n\\nℹ️lakeFS version {v}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1c3ca40-a690-4885-90f9-fc7c069c66c5",
   "metadata": {},
   "source": [
    "### Define lakeFS Repository"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "384aa4d3-038a-4a56-b30e-c536e3911478",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "repo = lakefs.Repository(repo_name).create(storage_namespace=f\"{storageNamespace}/{repo_name}\", default_branch=mainBranch, exist_ok=True)\n",
    "branchMain = repo.branch(mainBranch)\n",
    "print(repo)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f0a012b1-a70b-4d76-b433-94d9f4610a61",
   "metadata": {},
   "source": [
    "### Set up Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6e7196c9-370a-441a-8ea3-02ee03484991",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"lakeFS / Jupyter\") \\\n",
    "        .config(\"spark.hadoop.fs.s3.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\") \\\n",
    "        .config(\"spark.hadoop.fs.s3a.endpoint\", lakefsEndPoint) \\\n",
    "        .config(\"spark.hadoop.fs.s3a.path.style.access\", \"true\") \\\n",
    "        .config(\"spark.hadoop.fs.s3a.access.key\", lakefsAccessKey) \\\n",
    "        .config(\"spark.hadoop.fs.s3a.secret.key\", lakefsSecretKey) \\\n",
    "        .config(\"spark.jars.packages\", \"io.delta:delta-core_2.12:2.3.0\") \\\n",
    "        .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\") \\\n",
    "        .config(\"spark.sql.catalog.spark_catalog\", \"org.apache.spark.sql.delta.catalog.DeltaCatalog\") \\\n",
    "        .getOrCreate()\n",
    "spark.sparkContext.setLogLevel(\"INFO\")\n",
    "\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c757d0f4-929c-4d81-96de-7fa855d53fec",
   "metadata": {},
   "source": [
    "### Define a function to count records in a Delta table in different branches"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "044a35d3-7007-44d9-b2c8-cc57531a46d1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def delta_table_compare_branches(table, refs):\n",
    "  spark.createDataFrame(\n",
    "    data=zip(\n",
    "      refs,\n",
    "      map(lambda r: spark.read.format('delta').load(f's3a://{repo.id}/{r}/{table}').count(), refs)\n",
    "    ), \n",
    "    schema=StructType([ \n",
    "      StructField(\"Branch\", StringType(), True),\n",
    "      StructField(\"Count\", IntegerType(), True)\n",
    "    ])\n",
    "  ).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2377bae1-0cc2-4334-b36c-41fd17c93eba",
   "metadata": {},
   "source": [
    "### Define CUSTOMER.csv data file schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0e42f399-19b7-4def-9988-916a7f8ba067",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersSchema = StructType([\n",
    "  StructField(\"Customer_ID\", IntegerType(), False),\n",
    "  StructField(\"Country\", StringType(), False),\n",
    "  StructField(\"Gender\", StringType(), False),\n",
    "  StructField(\"Personal_ID\", IntegerType(), True),\n",
    "  StructField(\"Customer_Name\", StringType(), False),\n",
    "  StructField(\"Customer_FirstName\", StringType(), False),\n",
    "  StructField(\"Customer_LastName\", StringType(), False),\n",
    "  StructField(\"Birth_Date\", StringType(), False),\n",
    "  StructField(\"Customer_Address\", StringType(), False),\n",
    "  StructField(\"Street_ID\", LongType(), False),\n",
    "  StructField(\"Street_Number\", IntegerType(), False),\n",
    "  StructField(\"Customer_Type_ID\", IntegerType(), False)\n",
    "])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd6c69ac-32a2-4efa-bcc5-8bf5aa8fe867",
   "metadata": {},
   "source": [
    "### Define ORDER_FACT.csv data file schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6076bbcf-3bb2-4261-944a-0796304dea89",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersSchema = StructType([\n",
    "  StructField(\"Customer_ID\", IntegerType(), False),\n",
    "  StructField(\"Employee_ID\", IntegerType(), False),\n",
    "  StructField(\"Street_ID\", LongType(), False),\n",
    "  StructField(\"Order_Date\", StringType(), False),\n",
    "  StructField(\"Delivery_Date\", StringType(), False),\n",
    "  StructField(\"Order_ID\", LongType(), True),\n",
    "  StructField(\"Order_Type\", ByteType(), False),\n",
    "  StructField(\"Product_ID\", LongType(), False),\n",
    "  StructField(\"Quantity\", ByteType(), False),\n",
    "  StructField(\"Total_Retail_Price\", StringType(), False),\n",
    "  StructField(\"CostPrice_Per_Unit\", StringType(), False),\n",
    "  StructField(\"Discount\", LongType(), False)\n",
    "])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "25ea2edb-6c7e-46b9-9aea-fa15952cf3cb",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "406d9862-c34a-4281-80dc-32d15dc446b3",
   "metadata": {},
   "source": [
    "# Main demo starts here 🚦 👇🏻"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cbc87fa4-3d84-4c13-a612-b2f14d15d4e6",
   "metadata": {},
   "source": [
    "For this demo - we'll be utilizing a dataset - [Orion Star - Sports and outdoors RDBMS dataset](https://www.kaggle.com/datasets/chethanp11/orion-star-sports-and-outdoors-rdbms-dataset) from [Kaggle](https://www.kaggle.com/)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b1577257-81b2-4971-bafb-5aabfd1e603e",
   "metadata": {},
   "source": [
    "## Create Customers delta table in the main branch (using [CUSTOMER.csv](./data/samples/OrionStar/CUSTOMER.csv) file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7229f90d-accb-4f8f-9713-50cd7c8044b5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo.id}/{mainBranch}/{customersTable}\"\n",
    "print(customersTablePath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b3eecdfe-ecf0-4947-9feb-acfae738842f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = spark.read.csv('/data/OrionStar/CUSTOMER.csv',header=True,schema=customersSchema)\n",
    "df.write.format(\"delta\").mode(\"overwrite\").save(customersTablePath)\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "68e404d3-982f-48d3-a96c-b8bf566dab68",
   "metadata": {},
   "source": [
    "## Create Orders delta table in the main branch (using [ORDER_FACT.csv](./data/samples/OrionStar/ORDER_FACT.csv) file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "172e822c-efb9-4879-b8a6-dd6879f6cf23",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"s3a://{repo.id}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.csv('/data/OrionStar/ORDER_FACT.csv',header=True,schema=ordersSchema)\n",
    "df.write.format(\"delta\").mode(\"overwrite\").save(ordersTablePath)\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b763c2a9-561c-447c-ae2b-0b8fa13fedb0",
   "metadata": {},
   "source": [
    "## Commit changes and attach some metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "387aab4d-4f3f-4eeb-acc1-58fb6d32fd85",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ref = branchMain.commit(message='Added customers and orders Delta tables!', \n",
    "        metadata={'using': 'python_api'})\n",
    "print_commit(ref.get_commit())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "63ec287f-b340-43be-983f-fad43b024ac5",
   "metadata": {},
   "source": [
    "# 🟢 ETL Job Starts"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5fa6a1a9-70c2-4416-b860-97aa1c2a6dfa",
   "metadata": {},
   "source": [
    "## Create a new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1769a28d-2aec-445b-a5fd-a79b5ee723a8",
   "metadata": {
    "tags": []
   },
   "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",
   "id": "6a6e75ca-12c6-4aa9-81f2-e7e3e092a56b",
   "metadata": {},
   "source": [
    "## List the repository branches by using lakeFS Python client API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8b03960f-73ab-440e-9932-59d20757a806",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "for branchList in repo.branches():\n",
    "    print(branchList.id)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce2c7463-b3dd-43da-894b-25dbe92f8af6",
   "metadata": {},
   "source": [
    "## Apply POS (Point of Sale) Transactions to Delta Lake: delete data for a customer on the new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8228df43-a222-44b2-918d-2f4863717156",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from delta.tables import *\n",
    "ordersTablePath = f\"s3a://{repo.id}/{deltaLakeETLBranch}/{ordersTable}\"\n",
    "deltaTable = DeltaTable.forPath(spark, ordersTablePath)\n",
    "deltaTable.delete(\"Customer_ID = 19444\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46e19799-93b9-41a5-8764-d2bcc8d2e574",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo.id}/{deltaLakeETLBranch}/{customersTable}\"\n",
    "deltaTable = DeltaTable.forPath(spark, customersTablePath)\n",
    "deltaTable.delete(\"Customer_ID = 19444\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d889b50b-42ad-4aef-a13e-017c8342a68a",
   "metadata": {},
   "source": [
    "## Apply POS Transactions to Delta Lake: update data for a customer on the new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "65386bf4-5218-4fab-9db4-a983aad4320c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo.id}/{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",
   "id": "85a9b453-34e4-45be-8b07-59d7ee7f3514",
   "metadata": {},
   "source": [
    "## Apply POS Transactions to Delta Lake: batch upsert (5 updated and 10 new orders in [ORDER_FACT_UPDATES.csv](/data/samples/OrionStar/ORDER_FACT_UPDATES.csv) file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e804048d-1b14-4a06-8d31-31e3c4e7af08",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"s3a://{repo.id}/{deltaLakeETLBranch}/{ordersTable}\"\n",
    "deltaTableOrders = DeltaTable.forPath(spark, ordersTablePath)\n",
    "\n",
    "orderUpdatesTablePath = f\"s3a://{repo.id}/{deltaLakeETLBranch}/{orderUpdatesTable}\"\n",
    "dfOrderUpdates = spark.read.csv('/data/OrionStar/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",
   "id": "c32ca2cf-970d-4b16-9e4a-181f2bb88fe2",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Customers delta table in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "63499a12-9f23-465a-8655-75a3bb4ccc7e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo.id}/{mainBranch}/{customersTable}\"\n",
    "df = spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dd08a06a-fa95-4a52-a05e-30c602b0069d",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo.id}/{deltaLakeETLBranch}/{customersTable}\"\n",
    "spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4f31fb4-8bff-4b6d-87bf-3e8d520602c2",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Customers count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "227b3ac3-fcc6-4da0-b10b-64c5d8f9b1c4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(customersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f7dec0be-ee1e-471e-9c86-2ca2cc68f65c",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Orders delta table in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b1c5019-aca2-4592-ba68-f650b3513348",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"s3a://{repo.id}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "207ff757-f9ce-4df0-953a-5e8c0252b1eb",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"s3a://{repo.id}/{deltaLakeETLBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "da687f13-2d57-423d-b78f-f9e1e6f76bdf",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Orders count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0cb6bc56-f461-4717-a747-fda12328d017",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(ordersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3af4ccfe-823a-4f8c-9a6b-3f2bf520787c",
   "metadata": {},
   "source": [
    "## Commit changes and attach some metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "292622df-1c82-4221-ba46-d46455867725",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ref = branchDeltaLakeETL.commit(message='Deleted and updated customers. Deleted and upserted orders.', \n",
    "        metadata={'using': 'python_api'})\n",
    "print_commit(ref.get_commit())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b20e4afb-88c7-4a08-9ce8-c991be6f632d",
   "metadata": {},
   "source": [
    "## Diff between the new branch and the source branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f0d32244-70ee-4d95-bbb9-160851a689a7",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "diff = branchMain.diff(other_ref=branchDeltaLakeETL)\n",
    "print_diff(diff)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fe2d0f89-44c8-47bb-bcd9-39be49216b63",
   "metadata": {},
   "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": null,
   "id": "f161f740-c18b-4e7e-b9a2-ca08793f6a0c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Uncomment if you want to run this\n",
    "\n",
    "#branchDeltaLakeETL.delete()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "49a28f7a-bd23-4bfc-b3e7-12c35691f942",
   "metadata": {},
   "source": [
    "## Or merge new branch to the main branch if ETL job succeeds (atomic promotion to production)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ebdc00ab-72f3-4005-a90a-33dd28bc9c82",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "res = branchDeltaLakeETL.merge_into(branchMain)\n",
    "print(res)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "abc0f5c7-4eaf-4797-995d-e823e36e2654",
   "metadata": {},
   "source": [
    "## Data Validation: Read data from the main branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "31a0b706-af96-4737-b625-4900231f2e9f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo.id}/{mainBranch}/{customersTable}\"\n",
    "spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "65fa7bd5-9f76-4879-85d2-308473b982b8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"s3a://{repo.id}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4d036ad-5e80-4e11-8e56-87d2c25195ca",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Customers count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a1a1f00f-d247-4e31-97f8-7ae6efa5a76a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(customersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67b363c6-d3a5-4e20-b1f6-6ce078986a40",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Orders count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a917cb01-b720-42a2-bf1c-37be203389cd",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(ordersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "60260a17-0d22-4b33-a9fe-2bc8d4d0f6c1",
   "metadata": {},
   "source": [
    "## If you merged new branch to the main branch then you can atomically rollback Multi-Table Transactions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "13be9af7-5fcb-467c-9073-337bf7360e44",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "branchMain.revert(parent_number=1, reference=mainBranch)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a889314f-3e63-4a83-bb8e-af5f5f038129",
   "metadata": {},
   "source": [
    "## Data Validation: Read data again from the main branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b57615a0-4f6d-4cab-b62d-6f3f6300f1a4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo.id}/{mainBranch}/{customersTable}\"\n",
    "spark.read.format(\"delta\").load(customersTablePath).where(\"Customer_ID = 19444 OR Customer_ID = 63\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bf5d0968-7c85-406a-96b8-66a97335ce7e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"s3a://{repo.id}/{mainBranch}/{ordersTable}\"\n",
    "df = spark.read.format(\"delta\").load(ordersTablePath).where(\"Customer_ID = 19444\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2de098bc-c6bc-45d7-8bbe-f45d6055006c",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Customers count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d8a001aa-327d-4d4e-ac39-03bf73d2828f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(customersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5f96f041-4cc1-4172-9c14-f4fc910da1ef",
   "metadata": {},
   "source": [
    "## Data Validation: Compare Orders count in the main and new branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4189462d-364a-40ac-8b06-4590b8d8fad1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "refs = [mainBranch, deltaLakeETLBranch]\n",
    "\n",
    "delta_table_compare_branches(ordersTable, refs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "557472bd-39ed-43d3-a00a-ed69ed8ab280",
   "metadata": {},
   "source": [
    "## More Questions?\n",
    "\n",
    "###### Join the lakeFS Slack group - https://lakefs.io/slack"
   ]
  }
 ],
 "metadata": {
  "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}