Example_Data_Versioning_with_Delta_Lake.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "private_outputs": true,
      "provenance": [],
      "authorship_tag": "ABX9TyMAMfvNE09DpP39Lb36muw7",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/OneFineStarstuff/OneFineStardust/blob/main/Example_Data_Versioning_with_Delta_Lake.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "pip install delta"
      ],
      "metadata": {
        "id": "JE2aJEfuHCHh"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "pip install delta-spark"
      ],
      "metadata": {
        "id": "XkKe5NWvHb60"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from delta.tables import DeltaTable\n",
        "\n",
        "# Initialize Spark session with Delta Lake support\n",
        "spark = SparkSession.builder \\\n",
        "    .appName(\"DeltaTableExample\") \\\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",
        "\n",
        "# Check if the path exists before accessing the Delta table\n",
        "path = \"/path/to/data\"\n",
        "\n",
        "# Initialize DeltaTable if the path exists\n",
        "try:\n",
        "    deltaTable = DeltaTable.forPath(spark, path)\n",
        "\n",
        "    # Perform update operations and save new versions\n",
        "    deltaTable.update(\n",
        "        condition=\"id = 1\",\n",
        "        set={\"column_name\": \"'new_value'\"}\n",
        "    )\n",
        "\n",
        "    # Retrieve specific data version\n",
        "    historic_data = spark.read.format(\"delta\").option(\"versionAsOf\", 1).load(path)\n",
        "\n",
        "    # Show historic data\n",
        "    historic_data.show()\n",
        "\n",
        "except Exception as e:\n",
        "    print(f\"Error accessing Delta table at {path}: {e}\")"
      ],
      "metadata": {
        "id": "R_3IiLzeGuUp"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}