notebooks/pyspark/pyspark-ibis-delta-lake.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "186ddd52-4f3a-4127-b3f3-c20bcf1acdc2",
   "metadata": {},
   "source": [
    "# PySpark / Ibis / Delta Lake"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ddfb91a8-a232-4ce8-a1d6-13d6a1c7c45c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from delta import *\n",
    "import ibis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c9311a68-c80b-447d-8595-ca780b0c9c54",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "builder = (\n",
    "    pyspark.sql.SparkSession.builder.appName(\"MyApp\")\n",
    "    .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\")\n",
    "    .config(\n",
    "        \"spark.sql.catalog.spark_catalog\",\n",
    "        \"org.apache.spark.sql.delta.catalog.DeltaCatalog\",\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "072c9688-54b6-4da3-bdc8-ba5ca57870a7",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/Users/matthew.powers/opt/miniconda3/envs/pyspark-332-delta-230/lib/python3.9/site-packages/pyspark/jars/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /Users/matthew.powers/.ivy2/cache\n",
      "The jars for the packages stored in: /Users/matthew.powers/.ivy2/jars\n",
      "io.delta#delta-core_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-8e9c3584-51ed-4dc9-9484-30e53318fe4e;1.0\n",
      "\tconfs: [default]\n",
      "\tfound io.delta#delta-core_2.12;2.3.0 in central\n",
      "\tfound io.delta#delta-storage;2.3.0 in central\n",
      "\tfound org.antlr#antlr4-runtime;4.8 in central\n",
      ":: resolution report :: resolve 105ms :: artifacts dl 4ms\n",
      "\t:: modules in use:\n",
      "\tio.delta#delta-core_2.12;2.3.0 from central in [default]\n",
      "\tio.delta#delta-storage;2.3.0 from central in [default]\n",
      "\torg.antlr#antlr4-runtime;4.8 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   3   |   0   |   0   |   0   ||   3   |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-8e9c3584-51ed-4dc9-9484-30e53318fe4e\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 3 already retrieved (0kB/3ms)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/05/05 22:37:05 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "spark = configure_spark_with_delta_pip(builder).getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "91b0b011-c074-4553-96d1-6b85db1bd5f9",
   "metadata": {},
   "source": [
    "## Create a Delta table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "c3b6c26f-eeeb-4c97-a8d0-264402bd55a9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = spark.createDataFrame([(0, \"Bob\", 75), (1, \"Sue\", 25), (2, \"Jim\", 27)]).toDF(\n",
    "    \"id\", \"name\", \"age\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "f6c3bf90-1d40-4610-875b-559a75551bec",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/05/05 22:37:22 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df.write.format(\"delta\").save(\"tmp/fun_people\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f371557f-ec27-4c94-8c05-09f845ebb5cd",
   "metadata": {},
   "source": [
    "## Append to the Delta table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "cafa9cf5-097b-43ce-8068-9a36e077e6b5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = spark.createDataFrame([(8, \"Larry\", 19), (9, \"Jerry\", 69)]).toDF(\n",
    "    \"id\", \"name\", \"age\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "2d8ad2ad-ab1b-40f7-8b10-c2e15d06190b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df.write.format(\"delta\").mode(\"append\").save(\"tmp/fun_people\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f4de312a-6163-4de9-8930-ee367fd42411",
   "metadata": {},
   "source": [
    "## Create an Ibis table from the Delta table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "30fb7381-42c9-4013-b4dc-9012871afd2a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "con = ibis.pyspark.connect(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "938a7567-6e07-4a4c-87d8-737ebde0a4b1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark.read.format(\"delta\").load(\"tmp/fun_people\").createOrReplaceTempView(\"fun_people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "a76383e7-a736-4ae0-b6ed-a71d8fb182bd",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "table = con.table(\"fun_people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "d4fac24b-51b6-42c0-a8b2-03f29f7139fe",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "      <th>name</th>\n",
       "      <th>age</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>9</td>\n",
       "      <td>Jarry</td>\n",
       "      <td>69</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>8</td>\n",
       "      <td>Larry</td>\n",
       "      <td>75</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id   name  age\n",
       "0   9  Jarry   69\n",
       "1   8  Larry   75"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "table.filter(table.age >= 50).execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "ace917c5-93b6-43ac-b8d8-bc4f93c69926",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "      <th>name</th>\n",
       "      <th>age</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>9</td>\n",
       "      <td>Jarry</td>\n",
       "      <td>69</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>Sue</td>\n",
       "      <td>25</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id   name  age\n",
       "0   9  Jarry   69\n",
       "1   1    Sue   25"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "table.head(2).execute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8628c805-6b99-4ad8-9255-6269bdcb799d",
   "metadata": {},
   "source": [
    "## Time travel back to version 0 of the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "08d8de84-3e88-4452-bca6-6fdc7e574d1f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark.read.format(\"delta\").load(\"tmp/fun_people\").createOrReplaceTempView(\"fun_people_v0\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "0cb8435d-7a7f-4ca7-a84a-88cf167bc003",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "table_v0 = con.table(\"fun_people_v0\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "873dfa0d-d5ce-4115-af73-53c734501d50",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "      <th>name</th>\n",
       "      <th>age</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>9</td>\n",
       "      <td>Jarry</td>\n",
       "      <td>69</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>8</td>\n",
       "      <td>Larry</td>\n",
       "      <td>75</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id   name  age\n",
       "0   9  Jarry   69\n",
       "1   8  Larry   75"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "table_v0.filter(table_v0.age >= 50).execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "3a8edfc5-2cf1-4faa-801b-e40132f1e7e7",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "      <th>name</th>\n",
       "      <th>age</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>9</td>\n",
       "      <td>Jarry</td>\n",
       "      <td>69</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>Sue</td>\n",
       "      <td>25</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   id   name  age\n",
       "0   9  Jarry   69\n",
       "1   1    Sue   25"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "table_v0.head(2).execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ce7ab40-ebf9-4435-9a6c-5d6025ec4610",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3f11266a-a4b0-4850-8325-96ff88c755e4",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda env:pyspark-332-delta-230]",
   "language": "python",
   "name": "conda-env-pyspark-332-delta-230-py"
  },
  "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.9.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}