deltalake-with-emr-demo.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "fc2ab894-5619-4eb0-bacd-2174b01c1930",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T04:37:18.794860Z",
     "iopub.status.busy": "2022-01-31T04:37:18.794631Z",
     "iopub.status.idle": "2022-01-31T04:37:47.453460Z",
     "shell.execute_reply": "2022-01-31T04:37:47.452890Z",
     "shell.execute_reply.started": "2022-01-31T04:37:18.794832Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c7c5a9ccabe5428db4160b7d8bd18ab7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>2</td><td>application_1643358166445_0004</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-84-12.ec2.internal:20888/proxy/application_1643358166445_0004/\" >Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-88-238.ec2.internal:8042/node/containerlogs/container_1643358166445_0004_01_000001/livy\" >Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0f4a8617-6c86-46c3-a5fe-a6f4ad07fd02",
   "metadata": {},
   "source": [
    "### Adding Delta Jar package and configurations to the notebook\n",
    "\n",
    "Set up configuration with the correct version of Delta Lake that is compatible with Spark or PySpark.\n",
    "\n",
    "For more information, see [here](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/Deltausing-cluster-spark.html)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3fc88123",
   "metadata": {},
   "source": [
    "You need to configure `spark.jar.packages` according to the Delta version that matches your Spark version.\n",
    "\n",
    "For exmaple, if you are using Amazon EMR versions `emr-6.7.0`, you should use the first configurations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d7e0934d",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "%%configure -f\n",
    "\n",
    "{\n",
    "  \"conf\": {\n",
    "    \"spark.jars.packages\": \"io.delta:delta-core_2.12:1.2.1\",\n",
    "    \"spark.sql.extensions\": \"io.delta.sql.DeltaSparkSessionExtension\",\n",
    "    \"spark.sql.catalog.spark_catalog\": \"org.apache.spark.sql.delta.catalog.DeltaCatalog\"\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5902946d",
   "metadata": {},
   "source": [
    "If you are running Amazon EMR releases `>= emr-6.9.0` and `< emr-7.0.0`, you must also use the following configurations to use fine-grained access control based on Lake Formation with Delta Lake."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b37ffbb8",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "%%configure -f\n",
    "\n",
    "{\n",
    "  \"conf\": {\n",
    "    \"spark.jars.packages\": \"io.delta:delta-core_2.13:2.1.0\",\n",
    "    \"spark.sql.extensions\": \"io.delta.sql.DeltaSparkSessionExtension\",\n",
    "    \"spark.sql.catalog.spark_catalog\": \"org.apache.spark.sql.delta.catalog.DeltaCatalog\",\n",
    "    \"spark.sql.catalog.spark_catalog.lf.managed\": \"true\"\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "02b006f1",
   "metadata": {},
   "source": [
    "If you are running Amazon EMR releases `emr-7.x.x`, you must also use the following configurations to use fine-grained access control based on Lake Formation with Delta Lake."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "b009437e-7eb9-4b62-bcec-8e651d68f773",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:07:09.772468Z",
     "iopub.status.busy": "2022-01-31T06:07:09.772086Z",
     "iopub.status.idle": "2022-01-31T06:07:54.221617Z",
     "shell.execute_reply": "2022-01-31T06:07:54.220936Z",
     "shell.execute_reply.started": "2022-01-31T06:07:09.772442Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>4</td><td>application_1643358166445_0006</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-84-12.ec2.internal:20888/proxy/application_1643358166445_0006/\" >Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-88-238.ec2.internal:8042/node/containerlogs/container_1643358166445_0006_01_000001/livy\" >Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "Current session configs: <tt>{'conf': {'spark.jars.packages': 'io.delta:delta-core_2.12:1.2.1', 'spark.sql.extensions': 'io.delta.sql.DeltaSparkSessionExtension', 'spark.sql.catalog.spark_catalog': 'org.apache.spark.sql.delta.catalog.DeltaCatalog'}, 'proxyUser': 'user_xxx', 'kind': 'pyspark'}</tt><br>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>4</td><td>application_1643358166445_0006</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-84-12.ec2.internal:20888/proxy/application_1643358166445_0006/\" >Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-88-238.ec2.internal:8042/node/containerlogs/container_1643358166445_0006_01_000001/livy\" >Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%configure -f\n",
    "\n",
    "{\n",
    "  \"conf\": {\n",
    "    \"spark.jars.packages\": \"io.delta:delta-spark_2.13:3.1.0\",\n",
    "    \"spark.sql.extensions\": \"io.delta.sql.DeltaSparkSessionExtension\",\n",
    "    \"spark.sql.catalog.spark_catalog\": \"org.apache.spark.sql.delta.catalog.DeltaCatalog\",\n",
    "    \"spark.sql.catalog.spark_catalog.lf.managed\": \"true\"\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "50a93457-553a-4f98-b4a6-d04a362976a7",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:08:15.055953Z",
     "iopub.status.busy": "2022-01-31T06:08:15.055667Z",
     "iopub.status.idle": "2022-01-31T06:08:15.333394Z",
     "shell.execute_reply": "2022-01-31T06:08:15.332680Z",
     "shell.execute_reply.started": "2022-01-31T06:08:15.055924Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "d25c35926c9149818b76eef2b021255a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from delta.tables import *\n",
    "\n",
    "bucket = \"learn-deltalake-2022\" # Replace `learn-deltalake-2022` with your s3 path from here on.\n",
    "hive_table, spark_table = \"delta-hive-table\", \"delta-spark-table\"\n",
    "\n",
    "deltaHivePath = f\"s3a://{bucket}/{hive_table}/\"\n",
    "deltaSparkPath = f\"s3a://{bucket}/{spark_table}/\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ac16780b-6db3-4be2-af86-6c9538fe4e15",
   "metadata": {},
   "source": [
    "### Defining Methods to generate Json Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "08726009-d3e8-4d8c-9da8-317f36adc9ef",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T07:24:33.086563Z",
     "iopub.status.busy": "2022-01-31T07:24:33.086313Z",
     "iopub.status.idle": "2022-01-31T07:24:33.138008Z",
     "shell.execute_reply": "2022-01-31T07:24:33.137334Z",
     "shell.execute_reply.started": "2022-01-31T07:24:33.086537Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "dfe85ef88fea4d2f83163dcfb0ab16de",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from datetime import datetime\n",
    "\n",
    "def get_json_data(start, count, dest):\n",
    "    time_stamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')\n",
    "    data = [{\n",
    "        'trip_id': i,\n",
    "        'tstamp': time_stamp,\n",
    "        'route_id': chr(65 + (i % 10)),\n",
    "        'destination': dest[i % 10]\n",
    "    } for i in range(start, start + count)]\n",
    "    return data\n",
    "\n",
    "def create_json_df(spark, data):\n",
    "    sc = spark.sparkContext\n",
    "    return spark.read.json(sc.parallelize(data))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "728df1e4-6793-4a06-b8eb-b763938fac34",
   "metadata": {},
   "source": [
    "### Creating Spark dataframe from trip data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "a9a9a28e-77ee-481c-8683-4ac4215e9cea",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T07:24:39.045677Z",
     "iopub.status.busy": "2022-01-31T07:24:39.045355Z",
     "iopub.status.idle": "2022-01-31T07:24:48.356609Z",
     "shell.execute_reply": "2022-01-31T07:24:48.355698Z",
     "shell.execute_reply.started": "2022-01-31T07:24:39.045637Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cc9d368f1a7e44209a3547ad3a2f837c",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2000000\n",
      "+-------------+--------+-------+-------------------+\n",
      "|  destination|route_id|trip_id|             tstamp|\n",
      "+-------------+--------+-------+-------------------+\n",
      "|      Seattle|       A|      0|2022-01-31 07:24:39|\n",
      "|     New York|       B|      1|2022-01-31 07:24:39|\n",
      "|   New Jersey|       C|      2|2022-01-31 07:24:39|\n",
      "|  Los Angeles|       D|      3|2022-01-31 07:24:39|\n",
      "|    Las Vagas|       E|      4|2022-01-31 07:24:39|\n",
      "|       Tucson|       F|      5|2022-01-31 07:24:39|\n",
      "| Washinton DC|       G|      6|2022-01-31 07:24:39|\n",
      "| Philadelphia|       H|      7|2022-01-31 07:24:39|\n",
      "|        Miami|       I|      8|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|      9|2022-01-31 07:24:39|\n",
      "|      Seattle|       A|     10|2022-01-31 07:24:39|\n",
      "|     New York|       B|     11|2022-01-31 07:24:39|\n",
      "|   New Jersey|       C|     12|2022-01-31 07:24:39|\n",
      "|  Los Angeles|       D|     13|2022-01-31 07:24:39|\n",
      "|    Las Vagas|       E|     14|2022-01-31 07:24:39|\n",
      "|       Tucson|       F|     15|2022-01-31 07:24:39|\n",
      "| Washinton DC|       G|     16|2022-01-31 07:24:39|\n",
      "| Philadelphia|       H|     17|2022-01-31 07:24:39|\n",
      "|        Miami|       I|     18|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|     19|2022-01-31 07:24:39|\n",
      "+-------------+--------+-------+-------------------+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "dest = [\"Seattle\", \"New York\", \"New Jersey\", \"Los Angeles\", \"Las Vagas\",\n",
    "        \"Tucson\", \"Washinton DC\", \"Philadelphia\", \"Miami\", \"San Francisco\"]\n",
    "\n",
    "df1 = create_json_df(spark, get_json_data(0, 2000000, dest))\n",
    "print(df1.count())\n",
    "df1.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "157b9be7-d614-43e4-9482-7105667e8461",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:35:59.997083Z",
     "iopub.status.busy": "2022-01-31T05:35:59.996851Z",
     "iopub.status.idle": "2022-01-31T05:36:00.055217Z",
     "shell.execute_reply": "2022-01-31T05:36:00.054532Z",
     "shell.execute_reply.started": "2022-01-31T05:35:59.997059Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7d61458731184b89a4ea0ccee82af483",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- destination: string (nullable = true)\n",
      " |-- route_id: string (nullable = true)\n",
      " |-- trip_id: long (nullable = true)\n",
      " |-- tstamp: string (nullable = true)"
     ]
    }
   ],
   "source": [
    "df1.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "23ec5c00-f899-42fb-9bb1-96d120ab3a66",
   "metadata": {},
   "source": [
    "### Saving the data frame in Delta format in S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "50f6bf27-8549-4cd0-a942-77691a9580e5",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:47:59.217815Z",
     "iopub.status.busy": "2022-01-31T05:47:59.217579Z",
     "iopub.status.idle": "2022-01-31T05:48:08.542470Z",
     "shell.execute_reply": "2022-01-31T05:48:08.541841Z",
     "shell.execute_reply.started": "2022-01-31T05:47:59.217789Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e4f93ce9162e4be8b8493dfe2f8c06f1",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "df1.write.mode(\"overwrite\").format(\"delta\").partitionBy(\"route_id\").save(deltaHivePath)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5f8086f8-cf9b-414f-bed8-0a0a3be514e0",
   "metadata": {},
   "source": [
    "### Review written data using Spark SQL temporary view"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "53966c72-16a7-48f1-bfea-d072f33a90d4",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:48:15.049928Z",
     "iopub.status.busy": "2022-01-31T05:48:15.049698Z",
     "iopub.status.idle": "2022-01-31T05:48:22.402740Z",
     "shell.execute_reply": "2022-01-31T05:48:22.402051Z",
     "shell.execute_reply.started": "2022-01-31T05:48:15.049904Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "5b7fffdd0ddd4c30af6190b990ed32cf",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "| 2000000|\n",
      "+--------+\n",
      "\n",
      "+-------------+--------+-------+-------------------+\n",
      "|  destination|route_id|trip_id|             tstamp|\n",
      "+-------------+--------+-------+-------------------+\n",
      "|San Francisco|       J| 749569|2022-01-31 05:47:46|\n",
      "|San Francisco|       J| 749579|2022-01-31 05:47:46|\n",
      "|San Francisco|       J| 749589|2022-01-31 05:47:46|\n",
      "|San Francisco|       J| 749599|2022-01-31 05:47:46|\n",
      "|San Francisco|       J| 749609|2022-01-31 05:47:46|\n",
      "+-------------+--------+-------+-------------------+\n",
      "\n",
      "+------------+\n",
      "|max(trip_id)|\n",
      "+------------+\n",
      "|     1999999|\n",
      "+------------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "\n",
    "spark.sql(\"SELECT count(*) FROM temp_trip_table\").show()\n",
    "spark.sql(\"SELECT * FROM temp_trip_table LIMIT 5\").show()\n",
    "spark.sql(\"SELECT max(trip_id) FROM temp_trip_table\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3858cbb8-4236-416e-b463-b780cfc517c6",
   "metadata": {},
   "source": [
    "### Now lets create manifest files for accessing this delta storage data from Athena as table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "e1fd8652-a9c4-497c-838f-c30b4e127a17",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:48:26.378321Z",
     "iopub.status.busy": "2022-01-31T05:48:26.378064Z",
     "iopub.status.idle": "2022-01-31T05:48:31.687007Z",
     "shell.execute_reply": "2022-01-31T05:48:31.686379Z",
     "shell.execute_reply.started": "2022-01-31T05:48:26.378293Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "07891676ca6d4c639798f8391086ba0b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaTable = DeltaTable.forPath(spark, deltaHivePath)\n",
    "deltaTable.generate(\"symlink_format_manifest\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd3dbdd7",
   "metadata": {},
   "source": [
    "### Use Hive to Query Delta Lake tables in Amazon Athena\n",
    "\n",
    "Open the Athena console at https://console.aws.amazon.com/athena/, and run the sample queries at[`amazon_athena_queries_on_deltalake`](https://github.com/aws-samples/amazon-emr-with-delta-lake/blob/main/amazon_athena_queries_on_deltalake.md) in the Athena Query Editor.\n",
    "\n",
    "Then come back here, and run the rest fo the remaining Spark queries."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "42ea223b-2b25-4d67-8051-ef390c7b795f",
   "metadata": {},
   "source": [
    "### Creating a Spark table in Glue Data Catalog with same data\n",
    "\n",
    "Before running the following command, replace the s3 location with yours."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "b1095dff-09c7-4c02-b763-848dc8605205",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T07:25:45.774437Z",
     "iopub.status.busy": "2022-01-31T07:25:45.774180Z",
     "iopub.status.idle": "2022-01-31T07:25:45.831986Z",
     "shell.execute_reply": "2022-01-31T07:25:45.831361Z",
     "shell.execute_reply.started": "2022-01-31T07:25:45.774409Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%sql -q\n",
    "CREATE DATABASE IF NOT EXISTS my_delta_db LOCATION 's3a://learn-deltalake-2022/delta-spark-table/'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "d22d001d-027d-480f-950e-dbfc20b6bf60",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T07:25:57.220622Z",
     "iopub.status.busy": "2022-01-31T07:25:57.220389Z",
     "iopub.status.idle": "2022-01-31T07:25:57.525862Z",
     "shell.execute_reply": "2022-01-31T07:25:57.525131Z",
     "shell.execute_reply.started": "2022-01-31T07:25:57.220597Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e2ec0c2c6e5c4f49a60b181e78cba372",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaDatabase = \"my_delta_db\"\n",
    "deltaTable = f\"{deltaDatabase}.delta_spark_table\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "7b9cbb97-1a95-4535-9484-7b664748a0d4",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T07:26:07.210622Z",
     "iopub.status.busy": "2022-01-31T07:26:07.210389Z",
     "iopub.status.idle": "2022-01-31T07:26:20.525862Z",
     "shell.execute_reply": "2022-01-31T07:26:20.525131Z",
     "shell.execute_reply.started": "2022-01-31T07:26:07.210597Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "173624a9979c43dcb007a31067cef64c",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "df1.write.format(\"delta\").mode(\"overwrite\").partitionBy(\"route_id\").option(\"path\", deltaSparkPath).saveAsTable(deltaTable)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "b4cc3bf9-dd4e-4a35-8f0a-b1a2c45e1499",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:49:25.977422Z",
     "iopub.status.busy": "2022-01-31T05:49:25.977085Z",
     "iopub.status.idle": "2022-01-31T05:49:26.859801Z",
     "shell.execute_reply": "2022-01-31T05:49:26.856669Z",
     "shell.execute_reply.started": "2022-01-31T05:49:25.977380Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c2b2daff62364cef8aecb3a7d7b8f449",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------------+-----------+\n",
      "|   database|        tableName|isTemporary|\n",
      "+-----------+-----------------+-----------+\n",
      "|my_delta_db|delta_spark_table|      false|\n",
      "|my_delta_db| delta_trip_table|      false|\n",
      "|           |  temp_trip_table|       true|\n",
      "+-----------+-----------------+-----------+"
     ]
    }
   ],
   "source": [
    "spark.sql(f\"use {deltaDatabase}\")\n",
    "spark.sql(\"show tables\").show(20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "05e67a9f-b8b9-4b1e-b165-7fba030d5488",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T07:28:08.182675Z",
     "iopub.status.busy": "2022-01-31T07:28:08.182437Z",
     "iopub.status.idle": "2022-01-31T07:28:24.718996Z",
     "shell.execute_reply": "2022-01-31T07:28:24.718413Z",
     "shell.execute_reply.started": "2022-01-31T07:28:08.182650Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "| 2000000|\n",
      "+--------+\n",
      "\n",
      "+------------+\n",
      "|max(trip_id)|\n",
      "+------------+\n",
      "|     1999999|\n",
      "+------------+\n",
      "\n",
      "+-------------+--------+-------+-------------------+\n",
      "|  destination|route_id|trip_id|             tstamp|\n",
      "+-------------+--------+-------+-------------------+\n",
      "|San Francisco|       J|1750019|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750029|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750039|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750049|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750059|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750069|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750079|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750089|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750099|2022-01-31 07:24:39|\n",
      "|San Francisco|       J|1750109|2022-01-31 07:24:39|\n",
      "+-------------+--------+-------+-------------------+\n",
      "\n",
      "+--------+\n",
      "|route_id|\n",
      "+--------+\n",
      "|       A|\n",
      "|       B|\n",
      "|       C|\n",
      "|       D|\n",
      "|       E|\n",
      "|       F|\n",
      "|       G|\n",
      "|       H|\n",
      "|       I|\n",
      "|       J|\n",
      "+--------+"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%spark -c spark -o delta_trip_op\n",
    "\n",
    "delta_trip_op = spark.sql(f\"select count(*) from {deltaTable}\")\n",
    "delta_trip_op.show()\n",
    "\n",
    "delta_trip_op = spark.sql(f\"select max(trip_id) from {deltaTable}\")\n",
    "delta_trip_op.show()\n",
    "\n",
    "delta_trip_op = spark.sql(f\"select destination, route_id, trip_id, tstamp from {deltaTable} limit 10\")\n",
    "delta_trip_op.show()\n",
    "\n",
    "delta_trip_op = spark.sql(f\"select distinct route_id from {deltaTable} order by route_id\")\n",
    "delta_trip_op.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "faa0c5bb",
   "metadata": {},
   "source": [
    "### Querying Delta Lake tables in Amazon Athena\n",
    "\n",
    "Let's run queries on Delta Lake tables without using Hive at this point.\n",
    "\n",
    "Revisit the Athena console at https://console.aws.amazon.com/athena/, and run the sample queries at[`amazon_athena_queries_on_deltalake`](https://github.com/aws-samples/amazon-emr-with-delta-lake/blob/main/amazon_athena_queries_on_deltalake.md) in the Athena Query Editor.\n",
    "\n",
    "Then come back here, and run the rest fo the remaining Spark queries."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "038098f3-a776-4e7b-8d5e-978a9745157f",
   "metadata": {},
   "source": [
    "### This point forward only Athena table will be used for data processing"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3f156da7-e765-4127-900b-7eb4e5022ef9",
   "metadata": {},
   "source": [
    "### Lets INSERT some new rows to the Athena table with schema evolution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "02c4fa7b-d52a-4c9a-8a0b-c62356047f6f",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:31:31.501650Z",
     "iopub.status.busy": "2022-01-31T06:31:31.501416Z",
     "iopub.status.idle": "2022-01-31T06:31:31.562064Z",
     "shell.execute_reply": "2022-01-31T06:31:31.561365Z",
     "shell.execute_reply.started": "2022-01-31T06:31:31.501626Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "07d3d3d33b5a441bb92c64399f6e6f5a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from datetime import datetime\n",
    "\n",
    "def get_json_data_update(start, count, dest, origin):\n",
    "    time_stamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')\n",
    "    data = [{\n",
    "        'trip_id': i,\n",
    "        'tstamp': time_stamp,\n",
    "        'route_id': chr(65 + (i % 10)),\n",
    "        'destination': dest[i % 10],\n",
    "        'origin': origin[i % 10]\n",
    "    } for i in range(start, start + count)]\n",
    "    return data\n",
    "\n",
    "def create_json_df_update(spark, data):\n",
    "    sc = spark.sparkContext\n",
    "    return spark.read.json(sc.parallelize(data))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "52d353c6-63ed-4856-b37f-41d7dc3ac54a",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:49:39.886029Z",
     "iopub.status.busy": "2022-01-31T05:49:39.885809Z",
     "iopub.status.idle": "2022-01-31T05:49:40.655284Z",
     "shell.execute_reply": "2022-01-31T05:49:40.654630Z",
     "shell.execute_reply.started": "2022-01-31T05:49:39.886005Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "70119e5986b44324b13e233c39061263",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20\n",
      "+-----------+-------------+--------+-------+-------------------+\n",
      "|destination|       origin|route_id|trip_id|             tstamp|\n",
      "+-----------+-------------+--------+-------+-------------------+\n",
      "|   Syracuse|      Seattle|       A|2000000|2022-01-31 05:49:39|\n",
      "|   Syracuse|     New York|       B|2000001|2022-01-31 05:49:39|\n",
      "|   Syracuse|   New Jersey|       C|2000002|2022-01-31 05:49:39|\n",
      "|   Syracuse|  Los Angeles|       D|2000003|2022-01-31 05:49:39|\n",
      "|   Syracuse|    Las Vagas|       E|2000004|2022-01-31 05:49:39|\n",
      "|   Syracuse|       Tucson|       F|2000005|2022-01-31 05:49:39|\n",
      "|   Syracuse| Washinton DC|       G|2000006|2022-01-31 05:49:39|\n",
      "|   Syracuse| Philadelphia|       H|2000007|2022-01-31 05:49:39|\n",
      "|   Syracuse|        Miami|       I|2000008|2022-01-31 05:49:39|\n",
      "|   Syracuse|San Francisco|       J|2000009|2022-01-31 05:49:39|\n",
      "|   Syracuse|      Seattle|       A|2000010|2022-01-31 05:49:39|\n",
      "|   Syracuse|     New York|       B|2000011|2022-01-31 05:49:39|\n",
      "|   Syracuse|   New Jersey|       C|2000012|2022-01-31 05:49:39|\n",
      "|   Syracuse|  Los Angeles|       D|2000013|2022-01-31 05:49:39|\n",
      "|   Syracuse|    Las Vagas|       E|2000014|2022-01-31 05:49:39|\n",
      "|   Syracuse|       Tucson|       F|2000015|2022-01-31 05:49:39|\n",
      "|   Syracuse| Washinton DC|       G|2000016|2022-01-31 05:49:39|\n",
      "|   Syracuse| Philadelphia|       H|2000017|2022-01-31 05:49:39|\n",
      "|   Syracuse|        Miami|       I|2000018|2022-01-31 05:49:39|\n",
      "|   Syracuse|San Francisco|       J|2000019|2022-01-31 05:49:39|\n",
      "+-----------+-------------+--------+-------+-------------------+"
     ]
    }
   ],
   "source": [
    "insert_dest = ['Syracuse', 'Syracuse', 'Syracuse', 'Syracuse', 'Syracuse',\n",
    "               'Syracuse', 'Syracuse', 'Syracuse', 'Syracuse', 'Syracuse']\n",
    "insert_origin = [\"Seattle\", \"New York\", \"New Jersey\", \"Los Angeles\", \"Las Vagas\",\n",
    "                 \"Tucson\", \"Washinton DC\", \"Philadelphia\", \"Miami\", \"San Francisco\"]\n",
    "\n",
    "tripUpdates = create_json_df_update(spark, get_json_data_update(2000000, 20, insert_dest, insert_origin))\n",
    "print(tripUpdates.count())\n",
    "tripUpdates.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "3eb22325-7575-40b8-9af6-2f762e8b88f9",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:49:56.398247Z",
     "iopub.status.busy": "2022-01-31T05:49:56.397999Z",
     "iopub.status.idle": "2022-01-31T05:49:56.461891Z",
     "shell.execute_reply": "2022-01-31T05:49:56.461251Z",
     "shell.execute_reply.started": "2022-01-31T05:49:56.398221Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e7438433bb834ed4bcb0f7c720f1ac23",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- destination: string (nullable = true)\n",
      " |-- origin: string (nullable = true)\n",
      " |-- route_id: string (nullable = true)\n",
      " |-- trip_id: long (nullable = true)\n",
      " |-- tstamp: string (nullable = true)"
     ]
    }
   ],
   "source": [
    "tripUpdates.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "8f166b3c-de38-40d0-a9fa-25d28982a72b",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:50:14.590141Z",
     "iopub.status.busy": "2022-01-31T05:50:14.589912Z",
     "iopub.status.idle": "2022-01-31T05:50:17.870681Z",
     "shell.execute_reply": "2022-01-31T05:50:17.870053Z",
     "shell.execute_reply.started": "2022-01-31T05:50:14.590117Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "75d6c8d1ee8348ffa5bd0782ac451310",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaHivePath = f\"s3a://{bucket}/{hive_table}/\"\n",
    "tripUpdates.write.format(\"delta\").mode(\"append\").option(\"mergeSchema\", \"true\").partitionBy(\"route_id\").save(deltaHivePath)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a4bc4862-d08f-48ae-b074-fc48474abc77",
   "metadata": {},
   "source": [
    "### Now lets query the updated data using Spark SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "9c230268-39df-4719-92e7-a7d384d2e2ba",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:50:25.398200Z",
     "iopub.status.busy": "2022-01-31T05:50:25.397951Z",
     "iopub.status.idle": "2022-01-31T05:50:28.697593Z",
     "shell.execute_reply": "2022-01-31T05:50:28.696964Z",
     "shell.execute_reply.started": "2022-01-31T05:50:25.398173Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7344c95aaec148d3b03f849c38b75175",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "| 2000020|\n",
      "+--------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "spark.sql(\"SELECT count(*) FROM temp_trip_table\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "67798153-078a-4b83-9553-8eb432cc7f70",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:51:05.671041Z",
     "iopub.status.busy": "2022-01-31T05:51:05.670805Z",
     "iopub.status.idle": "2022-01-31T05:51:08.989801Z",
     "shell.execute_reply": "2022-01-31T05:51:08.988906Z",
     "shell.execute_reply.started": "2022-01-31T05:51:05.671016Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "8f6f37b91b854f70b14976d1a473242a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------+-------+-------------------+-------------+\n",
      "|  destination|route_id|trip_id|             tstamp|       origin|\n",
      "+-------------+--------+-------+-------------------+-------------+\n",
      "| Philadelphia|       H|1999997|2022-01-31 05:47:46|         null|\n",
      "|        Miami|       I|1999998|2022-01-31 05:47:46|         null|\n",
      "|San Francisco|       J|1999999|2022-01-31 05:47:46|         null|\n",
      "|     Syracuse|       A|2000000|2022-01-31 05:49:39|      Seattle|\n",
      "|     Syracuse|       B|2000001|2022-01-31 05:49:39|     New York|\n",
      "|     Syracuse|       C|2000002|2022-01-31 05:49:39|   New Jersey|\n",
      "|     Syracuse|       D|2000003|2022-01-31 05:49:39|  Los Angeles|\n",
      "|     Syracuse|       E|2000004|2022-01-31 05:49:39|    Las Vagas|\n",
      "|     Syracuse|       F|2000005|2022-01-31 05:49:39|       Tucson|\n",
      "|     Syracuse|       G|2000006|2022-01-31 05:49:39| Washinton DC|\n",
      "|     Syracuse|       H|2000007|2022-01-31 05:49:39| Philadelphia|\n",
      "|     Syracuse|       I|2000008|2022-01-31 05:49:39|        Miami|\n",
      "|     Syracuse|       J|2000009|2022-01-31 05:49:39|San Francisco|\n",
      "|     Syracuse|       A|2000010|2022-01-31 05:49:39|      Seattle|\n",
      "|     Syracuse|       B|2000011|2022-01-31 05:49:39|     New York|\n",
      "|     Syracuse|       C|2000012|2022-01-31 05:49:39|   New Jersey|\n",
      "|     Syracuse|       D|2000013|2022-01-31 05:49:39|  Los Angeles|\n",
      "|     Syracuse|       E|2000014|2022-01-31 05:49:39|    Las Vagas|\n",
      "|     Syracuse|       F|2000015|2022-01-31 05:49:39|       Tucson|\n",
      "|     Syracuse|       G|2000016|2022-01-31 05:49:39| Washinton DC|\n",
      "+-------------+--------+-------+-------------------+-------------+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM temp_trip_table WHERE trip_id > 1999996 ORDER BY trip_id\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73be88ed-112c-476e-993e-24287644e749",
   "metadata": {},
   "source": [
    "### Now Updating the Manifest files for Athena"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "747d65bb-c75c-420e-8613-62175af18fc1",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:52:00.430279Z",
     "iopub.status.busy": "2022-01-31T05:52:00.430049Z",
     "iopub.status.idle": "2022-01-31T05:52:03.731810Z",
     "shell.execute_reply": "2022-01-31T05:52:03.731173Z",
     "shell.execute_reply.started": "2022-01-31T05:52:00.430255Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "6a5c12109e2b450da50e8b7bc232614f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaTable = DeltaTable.forPath(spark, deltaHivePath)\n",
    "deltaTable.generate(\"symlink_format_manifest\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b1d6187f-c3fa-4f2f-a642-bd09b3c00b64",
   "metadata": {},
   "source": [
    "### Now lets UPDATE some data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "097e2223-1d13-45f7-b7a6-0ab00f8ab1e6",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T05:59:20.586612Z",
     "iopub.status.busy": "2022-01-31T05:59:20.586382Z",
     "iopub.status.idle": "2022-01-31T05:59:27.877624Z",
     "shell.execute_reply": "2022-01-31T05:59:27.877006Z",
     "shell.execute_reply.started": "2022-01-31T05:59:20.586588Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "4e5df906b1c64b8299b0de1584d4f0b3",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|Count_Syracuse|\n",
      "+--------------+\n",
      "|            20|\n",
      "+--------------+\n",
      "\n",
      "+------------------+\n",
      "|Count_Philadelphia|\n",
      "+------------------+\n",
      "|            200000|\n",
      "+------------------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "spark.sql(\"SELECT count(*) AS Count_Syracuse FROM temp_trip_table WHERE destination = 'Syracuse'\").show()\n",
    "spark.sql(\"SELECT count(*) AS Count_Philadelphia FROM temp_trip_table WHERE destination = 'Philadelphia'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "e33cddd0-db75-494d-a992-e1243ba89dd1",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:08:25.147745Z",
     "iopub.status.busy": "2022-01-31T06:08:25.147523Z",
     "iopub.status.idle": "2022-01-31T06:08:54.536682Z",
     "shell.execute_reply": "2022-01-31T06:08:54.535939Z",
     "shell.execute_reply.started": "2022-01-31T06:08:25.147720Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "56646d3bedbe4cde9a5ecaed2acfee82",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaTable = DeltaTable.forPath(spark, deltaHivePath)\n",
    "deltaTable.update(\"destination = 'Syracuse'\", {\"destination\": \"'Philadelphia'\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "a31675b2-2940-4976-bf20-da322a2d3fe2",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:09:13.638299Z",
     "iopub.status.busy": "2022-01-31T06:09:13.638053Z",
     "iopub.status.idle": "2022-01-31T06:09:22.984483Z",
     "shell.execute_reply": "2022-01-31T06:09:22.983642Z",
     "shell.execute_reply.started": "2022-01-31T06:09:13.638273Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c72adb9b99dc4734862b73927f0ae2c8",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaTable = DeltaTable.forPath(spark, deltaHivePath)\n",
    "deltaTable.generate(\"symlink_format_manifest\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "44af7816-3268-492d-a4c1-d56110d918fe",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:09:31.636001Z",
     "iopub.status.busy": "2022-01-31T06:09:31.635744Z",
     "iopub.status.idle": "2022-01-31T06:09:38.946201Z",
     "shell.execute_reply": "2022-01-31T06:09:38.945491Z",
     "shell.execute_reply.started": "2022-01-31T06:09:31.635976Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "295041bac16347b9917f1ad756c6cf7d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|Count_Syracuse|\n",
      "+--------------+\n",
      "|             0|\n",
      "+--------------+\n",
      "\n",
      "+------------------+\n",
      "|Count_Philadelphia|\n",
      "+------------------+\n",
      "|            200020|\n",
      "+------------------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "spark.sql(\"SELECT count(*) AS Count_Syracuse FROM temp_trip_table WHERE destination = 'Syracuse'\").show()\n",
    "spark.sql(\"SELECT count(*) AS Count_Philadelphia FROM temp_trip_table WHERE destination = 'Philadelphia'\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73422e6a-8b6c-44a7-8fc0-466e12dd762d",
   "metadata": {},
   "source": [
    "### Now lets DELETE some data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "74b0123c-47d5-46c5-88c5-f3ec32b79708",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:15:04.064199Z",
     "iopub.status.busy": "2022-01-31T06:15:04.063956Z",
     "iopub.status.idle": "2022-01-31T06:15:11.370838Z",
     "shell.execute_reply": "2022-01-31T06:15:11.370236Z",
     "shell.execute_reply.started": "2022-01-31T06:15:04.064174Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "04f4f4cdb17a4f348e085670a0993f27",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|Count_total|\n",
      "+-----------+\n",
      "|    2000020|\n",
      "+-----------+\n",
      "\n",
      "+--------+\n",
      "|Count_NJ|\n",
      "+--------+\n",
      "|  200000|\n",
      "+--------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "spark.sql(\"SELECT count(*) as Count_total FROM temp_trip_table\").show()\n",
    "spark.sql(\"SELECT count(*) as Count_NJ FROM temp_trip_table WHERE destination = 'New Jersey'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "815cc61d-75b8-4fea-be53-0f69d99a6ae7",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:16:01.728434Z",
     "iopub.status.busy": "2022-01-31T06:16:01.728194Z",
     "iopub.status.idle": "2022-01-31T06:16:09.062175Z",
     "shell.execute_reply": "2022-01-31T06:16:09.061230Z",
     "shell.execute_reply.started": "2022-01-31T06:16:01.728408Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "065c6db25a654b02812a92049a6a2952",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaTable = DeltaTable.forPath(spark, deltaHivePath)\n",
    "deltaTable.delete(\"destination = 'New Jersey'\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "5c6608d2-c8ff-4ad7-af01-12c6f8714ec6",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:20:06.044268Z",
     "iopub.status.busy": "2022-01-31T06:20:06.044036Z",
     "iopub.status.idle": "2022-01-31T06:20:11.345094Z",
     "shell.execute_reply": "2022-01-31T06:20:11.344191Z",
     "shell.execute_reply.started": "2022-01-31T06:20:06.044243Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "2efd58389f52421f8c5087b80daba6d6",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaTable.generate(\"symlink_format_manifest\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "55c3c5a4-f88d-40f0-8a73-eec94bfc2752",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:16:32.784037Z",
     "iopub.status.busy": "2022-01-31T06:16:32.783754Z",
     "iopub.status.idle": "2022-01-31T06:16:38.080838Z",
     "shell.execute_reply": "2022-01-31T06:16:38.080134Z",
     "shell.execute_reply.started": "2022-01-31T06:16:32.784008Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "69400b655b874a35911b308b4e1a0ff6",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|Count_total|\n",
      "+-----------+\n",
      "|    1800020|\n",
      "+-----------+\n",
      "\n",
      "+--------+\n",
      "|Count_NJ|\n",
      "+--------+\n",
      "|       0|\n",
      "+--------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "spark.sql(\"SELECT count(*) as Count_total FROM temp_trip_table\").show()\n",
    "spark.sql(\"SELECT count(*) as Count_NJ FROM temp_trip_table WHERE destination = 'New Jersey'\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "69bbc84a-e81a-4847-be63-9f65a8af0955",
   "metadata": {},
   "source": [
    "### Lets do some UPSERT now"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "16a2839b-d418-4d0b-9c82-ef1088141e19",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:17:26.632160Z",
     "iopub.status.busy": "2022-01-31T06:17:26.631930Z",
     "iopub.status.idle": "2022-01-31T06:17:29.929804Z",
     "shell.execute_reply": "2022-01-31T06:17:29.929062Z",
     "shell.execute_reply.started": "2022-01-31T06:17:26.632137Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cc5802be9c8447978c0b306ad80a1538",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+\n",
      "|max(trip_id)|\n",
      "+------------+\n",
      "|     2000019|\n",
      "+------------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "spark.sql(\"SELECT max(trip_id) FROM temp_trip_table\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "ee263fb5-d386-4392-8309-5d05e5d68279",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:18:23.236388Z",
     "iopub.status.busy": "2022-01-31T06:18:23.236157Z",
     "iopub.status.idle": "2022-01-31T06:18:26.518406Z",
     "shell.execute_reply": "2022-01-31T06:18:26.517774Z",
     "shell.execute_reply.started": "2022-01-31T06:18:23.236362Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "63c9509d2d064ce99c8116c240034680",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------+-------+-------------------+-------------+\n",
      "| destination|route_id|trip_id|             tstamp|       origin|\n",
      "+------------+--------+-------+-------------------+-------------+\n",
      "|Philadelphia|       F|2000015|2022-01-31 05:49:39|       Tucson|\n",
      "|Philadelphia|       G|2000016|2022-01-31 05:49:39| Washinton DC|\n",
      "|Philadelphia|       H|2000017|2022-01-31 05:49:39| Philadelphia|\n",
      "|Philadelphia|       I|2000018|2022-01-31 05:49:39|        Miami|\n",
      "|Philadelphia|       J|2000019|2022-01-31 05:49:39|San Francisco|\n",
      "+------------+--------+-------+-------------------+-------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM temp_trip_table WHERE trip_id > 2000014 ORDER BY trip_id\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "f185ef9d-54aa-4d70-8ea0-06379b89c87d",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:32:30.809562Z",
     "iopub.status.busy": "2022-01-31T06:32:30.809332Z",
     "iopub.status.idle": "2022-01-31T06:32:31.579059Z",
     "shell.execute_reply": "2022-01-31T06:32:31.578416Z",
     "shell.execute_reply.started": "2022-01-31T06:32:30.809537Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "31dc9b63248b4d73a46a5cbc5de3f01f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n",
      "+-----------+-------+--------+-------+-------------------+\n",
      "|destination| origin|route_id|trip_id|             tstamp|\n",
      "+-----------+-------+--------+-------+-------------------+\n",
      "|     Vernon|Seattle|       F|2000015|2022-01-31 06:32:30|\n",
      "|    Winsted|Seattle|       G|2000016|2022-01-31 06:32:30|\n",
      "|   Westport|Seattle|       H|2000017|2022-01-31 06:32:30|\n",
      "|  Fairfield|Seattle|       I|2000018|2022-01-31 06:32:30|\n",
      "|   Stamford|Seattle|       J|2000019|2022-01-31 06:32:30|\n",
      "|       Avon|Seattle|       A|2000020|2022-01-31 06:32:30|\n",
      "|   Simsbury|Seattle|       B|2000021|2022-01-31 06:32:30|\n",
      "| Farmington|Seattle|       C|2000022|2022-01-31 06:32:30|\n",
      "|    Windsor|Seattle|       D|2000023|2022-01-31 06:32:30|\n",
      "|  Ellington|Seattle|       E|2000024|2022-01-31 06:32:30|\n",
      "+-----------+-------+--------+-------+-------------------+"
     ]
    }
   ],
   "source": [
    "insert_dest = ['Avon', 'Simsbury', 'Farmington', 'Windsor', 'Ellington',\n",
    "               'Vernon', 'Winsted', 'Westport', 'Fairfield', 'Stamford']\n",
    "insert_origin = [\"Seattle\", \"Seattle\", \"Seattle\", \"Seattle\", \"Seattle\",\n",
    "                 \"Seattle\", \"Seattle\", \"Seattle\", \"Seattle\", \"Seattle\"]\n",
    "\n",
    "\n",
    "tripUpdates1 = create_json_df_update(spark, get_json_data_update(2000015, 10, insert_dest, insert_origin))\n",
    "print(tripUpdates1.count())\n",
    "tripUpdates1.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "c887989e-65e7-48b8-8d33-c8cc1906e20d",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:35:01.013408Z",
     "iopub.status.busy": "2022-01-31T06:35:01.013178Z",
     "iopub.status.idle": "2022-01-31T06:35:14.780401Z",
     "shell.execute_reply": "2022-01-31T06:35:14.779623Z",
     "shell.execute_reply.started": "2022-01-31T06:35:01.013383Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "714ff36311894016ad276743d371e582",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "deltaTable = DeltaTable.forPath(spark, deltaHivePath)\n",
    "(deltaTable\n",
    ".alias('t')\n",
    ".merge(tripUpdates1.alias('s'), 't.trip_id = s.trip_id')\n",
    ".whenMatchedUpdateAll()\n",
    ".whenNotMatchedInsertAll()\n",
    ".execute())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "365dcc5e-dcdd-4cff-bdc3-a657207175cc",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:36:24.397868Z",
     "iopub.status.busy": "2022-01-31T06:36:24.397629Z",
     "iopub.status.idle": "2022-01-31T06:36:27.679361Z",
     "shell.execute_reply": "2022-01-31T06:36:27.678504Z",
     "shell.execute_reply.started": "2022-01-31T06:36:24.397840Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e7b596e7064546bdb5f2774d96fcaae7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------+-------+-------------------+-----------+\n",
      "| destination|route_id|trip_id|             tstamp|     origin|\n",
      "+------------+--------+-------+-------------------+-----------+\n",
      "|Philadelphia|       B|2000011|2022-01-31 05:49:39|   New York|\n",
      "|Philadelphia|       C|2000012|2022-01-31 05:49:39| New Jersey|\n",
      "|Philadelphia|       D|2000013|2022-01-31 05:49:39|Los Angeles|\n",
      "|Philadelphia|       E|2000014|2022-01-31 05:49:39|  Las Vagas|\n",
      "|      Vernon|       F|2000015|2022-01-31 06:32:30|    Seattle|\n",
      "|     Winsted|       G|2000016|2022-01-31 06:32:30|    Seattle|\n",
      "|    Westport|       H|2000017|2022-01-31 06:32:30|    Seattle|\n",
      "|   Fairfield|       I|2000018|2022-01-31 06:32:30|    Seattle|\n",
      "|    Stamford|       J|2000019|2022-01-31 06:32:30|    Seattle|\n",
      "|        Avon|       A|2000020|2022-01-31 06:32:30|    Seattle|\n",
      "|    Simsbury|       B|2000021|2022-01-31 06:32:30|    Seattle|\n",
      "|  Farmington|       C|2000022|2022-01-31 06:32:30|    Seattle|\n",
      "|     Windsor|       D|2000023|2022-01-31 06:32:30|    Seattle|\n",
      "|   Ellington|       E|2000024|2022-01-31 06:32:30|    Seattle|\n",
      "+------------+--------+-------+-------------------+-----------+"
     ]
    }
   ],
   "source": [
    "spark.read.format(\"delta\").load(deltaHivePath).createOrReplaceTempView(\"temp_trip_table\")\n",
    "spark.sql(\"SELECT * FROM temp_trip_table WHERE trip_id > 2000010 ORDER BY trip_id\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2b676d92-f2fe-4304-a565-a7cb6128d1e5",
   "metadata": {},
   "source": [
    "### Lets perform some time travel queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "5950c8d6-7ab9-47cd-8e1d-cbe09f21ebf8",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:39:12.706196Z",
     "iopub.status.busy": "2022-01-31T06:39:12.705867Z",
     "iopub.status.idle": "2022-01-31T06:39:13.992985Z",
     "shell.execute_reply": "2022-01-31T06:39:13.992182Z",
     "shell.execute_reply.started": "2022-01-31T06:39:12.706154Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "79554819cc06409f9ed4c10c91e1129a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------------------+---------+---------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|version|timestamp          |operation|operationParameters                                                                                                                          |\n",
      "+-------+-------------------+---------+---------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|4      |2022-01-31 06:35:13|MERGE    |[predicate -> (t.`trip_id` = s.`trip_id`), matchedPredicates -> [{\"actionType\":\"update\"}], notMatchedPredicates -> [{\"actionType\":\"insert\"}]]|\n",
      "|3      |2022-01-31 06:16:07|DELETE   |[predicate -> [\"(`destination` = 'New Jersey')\"]]                                                                                            |\n",
      "|2      |2022-01-31 06:08:53|UPDATE   |[predicate -> (destination#216 = Syracuse)]                                                                                                  |\n",
      "|1      |2022-01-31 05:50:17|WRITE    |[mode -> Append, partitionBy -> [\"route_id\"]]                                                                                                |\n",
      "|0      |2022-01-31 05:48:06|WRITE    |[mode -> Overwrite, partitionBy -> [\"route_id\"]]                                                                                             |\n",
      "+-------+-------------------+---------+---------------------------------------------------------------------------------------------------------------------------------------------+"
     ]
    }
   ],
   "source": [
    "(deltaTable\n",
    ".history(100)\n",
    ".select(\"version\", \"timestamp\", \"operation\", \"operationParameters\")\n",
    ".show(truncate=False))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "205a828d-e56c-4668-bade-b2d146b5f927",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:41:18.010367Z",
     "iopub.status.busy": "2022-01-31T06:41:18.010119Z",
     "iopub.status.idle": "2022-01-31T06:41:21.302664Z",
     "shell.execute_reply": "2022-01-31T06:41:21.301945Z",
     "shell.execute_reply.started": "2022-01-31T06:41:18.010339Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cb902eb9b3884ed2a6dd1c4a24902477",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(destination='San Francisco', route_id='J', trip_id=749569, tstamp='2022-01-31 05:47:46'), Row(destination='San Francisco', route_id='J', trip_id=749579, tstamp='2022-01-31 05:47:46'), Row(destination='San Francisco', route_id='J', trip_id=749589, tstamp='2022-01-31 05:47:46'), Row(destination='San Francisco', route_id='J', trip_id=749599, tstamp='2022-01-31 05:47:46'), Row(destination='San Francisco', route_id='J', trip_id=749609, tstamp='2022-01-31 05:47:46')]"
     ]
    }
   ],
   "source": [
    "df_v0 = (spark.read\n",
    "         .format(\"delta\")\n",
    "         .option(\"timestampAsOf\", \"2022-01-31 05:48:06\") # Use the timestamp of version 0 for `timestampAsOf` parameter\n",
    "         .load(deltaHivePath))\n",
    "df_v0.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "76002f30-4508-45ab-abb4-21c319f7dd04",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:43:02.094258Z",
     "iopub.status.busy": "2022-01-31T06:43:02.094028Z",
     "iopub.status.idle": "2022-01-31T06:43:05.419081Z",
     "shell.execute_reply": "2022-01-31T06:43:05.418346Z",
     "shell.execute_reply.started": "2022-01-31T06:43:02.094233Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7ef08994f5994b6ab91949cdec660064",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------+-------+------+\n",
      "|destination|route_id|trip_id|tstamp|\n",
      "+-----------+--------+-------+------+\n",
      "+-----------+--------+-------+------+"
     ]
    }
   ],
   "source": [
    "df_v0.createOrReplaceTempView(\"temp_trip_table_v0\")\n",
    "spark.sql(\"SELECT * FROM temp_trip_table_v0 WHERE trip_id >= 2000000 ORDER BY trip_id\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "69b36557-6e49-4fdb-a207-98756cf3766a",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:46:10.855227Z",
     "iopub.status.busy": "2022-01-31T06:46:10.855000Z",
     "iopub.status.idle": "2022-01-31T06:46:13.134290Z",
     "shell.execute_reply": "2022-01-31T06:46:13.133586Z",
     "shell.execute_reply.started": "2022-01-31T06:46:10.855203Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "1808b03ec09b4343b5b69f8e3449a651",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(destination='San Francisco', route_id='J', trip_id=749569, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749579, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749589, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749599, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749609, tstamp='2022-01-31 05:47:46', origin=None)]"
     ]
    }
   ],
   "source": [
    "df_v2 = (spark.read\n",
    "         .format(\"delta\")\n",
    "         .option(\"versionAsOf\", \"2\")\n",
    "         .load(deltaHivePath))\n",
    "df_v2.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "260a8908-370a-4edb-a30b-f574ee3329ca",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:46:58.694581Z",
     "iopub.status.busy": "2022-01-31T06:46:58.694353Z",
     "iopub.status.idle": "2022-01-31T06:47:01.993791Z",
     "shell.execute_reply": "2022-01-31T06:47:01.992901Z",
     "shell.execute_reply.started": "2022-01-31T06:46:58.694557Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f00a26b28ec44454b09b2e90056a2678",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------+-------+-------------------+-------------+\n",
      "| destination|route_id|trip_id|             tstamp|       origin|\n",
      "+------------+--------+-------+-------------------+-------------+\n",
      "|Philadelphia|       A|2000000|2022-01-31 05:49:39|      Seattle|\n",
      "|Philadelphia|       B|2000001|2022-01-31 05:49:39|     New York|\n",
      "|Philadelphia|       C|2000002|2022-01-31 05:49:39|   New Jersey|\n",
      "|Philadelphia|       D|2000003|2022-01-31 05:49:39|  Los Angeles|\n",
      "|Philadelphia|       E|2000004|2022-01-31 05:49:39|    Las Vagas|\n",
      "|Philadelphia|       F|2000005|2022-01-31 05:49:39|       Tucson|\n",
      "|Philadelphia|       G|2000006|2022-01-31 05:49:39| Washinton DC|\n",
      "|Philadelphia|       H|2000007|2022-01-31 05:49:39| Philadelphia|\n",
      "|Philadelphia|       I|2000008|2022-01-31 05:49:39|        Miami|\n",
      "|Philadelphia|       J|2000009|2022-01-31 05:49:39|San Francisco|\n",
      "|Philadelphia|       A|2000010|2022-01-31 05:49:39|      Seattle|\n",
      "|Philadelphia|       B|2000011|2022-01-31 05:49:39|     New York|\n",
      "|Philadelphia|       C|2000012|2022-01-31 05:49:39|   New Jersey|\n",
      "|Philadelphia|       D|2000013|2022-01-31 05:49:39|  Los Angeles|\n",
      "|Philadelphia|       E|2000014|2022-01-31 05:49:39|    Las Vagas|\n",
      "|Philadelphia|       F|2000015|2022-01-31 05:49:39|       Tucson|\n",
      "|Philadelphia|       G|2000016|2022-01-31 05:49:39| Washinton DC|\n",
      "|Philadelphia|       H|2000017|2022-01-31 05:49:39| Philadelphia|\n",
      "|Philadelphia|       I|2000018|2022-01-31 05:49:39|        Miami|\n",
      "|Philadelphia|       J|2000019|2022-01-31 05:49:39|San Francisco|\n",
      "+------------+--------+-------+-------------------+-------------+"
     ]
    }
   ],
   "source": [
    "df_v2.createOrReplaceTempView(\"temp_trip_table_v2\")\n",
    "spark.sql(\"SELECT * FROM temp_trip_table_v2 WHERE trip_id >= 2000000 ORDER BY trip_id\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "8b103a96-d348-4ac5-a6c2-7f475093292c",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:47:58.970478Z",
     "iopub.status.busy": "2022-01-31T06:47:58.970252Z",
     "iopub.status.idle": "2022-01-31T06:48:02.258745Z",
     "shell.execute_reply": "2022-01-31T06:48:02.258158Z",
     "shell.execute_reply.started": "2022-01-31T06:47:58.970454Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "8d8cc8f6475c42a09a7e3c96e0504d32",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(destination='San Francisco', route_id='J', trip_id=749569, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749579, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749589, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749599, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749609, tstamp='2022-01-31 05:47:46', origin=None)]"
     ]
    }
   ],
   "source": [
    "df_v1 = (spark.read\n",
    "         .format(\"delta\")\n",
    "         .option(\"versionAsOf\", \"1\")\n",
    "         .load(deltaHivePath))\n",
    "df_v1.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "3c363b01-d7f4-4e97-9013-d501303f0312",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:48:18.626988Z",
     "iopub.status.busy": "2022-01-31T06:48:18.626760Z",
     "iopub.status.idle": "2022-01-31T06:48:20.931652Z",
     "shell.execute_reply": "2022-01-31T06:48:20.930893Z",
     "shell.execute_reply.started": "2022-01-31T06:48:18.626963Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7fc371d968d7474da9a4ab80c282d57d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(destination='San Francisco', route_id='J', trip_id=749569, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749579, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749589, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749599, tstamp='2022-01-31 05:47:46', origin=None), Row(destination='San Francisco', route_id='J', trip_id=749609, tstamp='2022-01-31 05:47:46', origin=None)]"
     ]
    }
   ],
   "source": [
    "df_v3 = (spark.read\n",
    "         .format(\"delta\")\n",
    "         .option(\"versionAsOf\", \"3\")\n",
    "         .load(deltaHivePath))\n",
    "df_v3.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "52214754-3d17-43f4-bd05-fcf094c289ae",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:49:53.566723Z",
     "iopub.status.busy": "2022-01-31T06:49:53.566491Z",
     "iopub.status.idle": "2022-01-31T06:50:00.882904Z",
     "shell.execute_reply": "2022-01-31T06:50:00.881859Z",
     "shell.execute_reply.started": "2022-01-31T06:49:53.566698Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cd457bf264af4fa3821179fdaa40ea77",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|Count_Syracuse|\n",
      "+--------------+\n",
      "|            20|\n",
      "+--------------+\n",
      "\n",
      "+------------------+\n",
      "|Count_Philadelphia|\n",
      "+------------------+\n",
      "|            200000|\n",
      "+------------------+"
     ]
    }
   ],
   "source": [
    "df_v1.createOrReplaceTempView(\"temp_trip_table_v1\")\n",
    "spark.sql(\"SELECT count(*) AS Count_Syracuse FROM temp_trip_table_v1 WHERE destination = 'Syracuse'\").show()\n",
    "spark.sql(\"SELECT count(*) AS Count_Philadelphia FROM temp_trip_table_v1 WHERE destination = 'Philadelphia'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "4aaabc08-ae99-4731-904b-7d49e9308315",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:50:41.434802Z",
     "iopub.status.busy": "2022-01-31T06:50:41.434575Z",
     "iopub.status.idle": "2022-01-31T06:50:46.725511Z",
     "shell.execute_reply": "2022-01-31T06:50:46.724909Z",
     "shell.execute_reply.started": "2022-01-31T06:50:41.434778Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "3063d1a710dd4c9eadf5d6d9abecab6f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|Count_Syracuse|\n",
      "+--------------+\n",
      "|             0|\n",
      "+--------------+\n",
      "\n",
      "+------------------+\n",
      "|Count_Philadelphia|\n",
      "+------------------+\n",
      "|            200020|\n",
      "+------------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT count(*) AS Count_Syracuse FROM temp_trip_table_v2 WHERE destination = 'Syracuse'\").show()\n",
    "spark.sql(\"SELECT count(*) AS Count_Philadelphia FROM temp_trip_table_v2 WHERE destination = 'Philadelphia'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "8b1a5e26-a73c-446a-8079-b63d2e176d19",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:52:38.995088Z",
     "iopub.status.busy": "2022-01-31T06:52:38.994854Z",
     "iopub.status.idle": "2022-01-31T06:52:44.288962Z",
     "shell.execute_reply": "2022-01-31T06:52:44.288114Z",
     "shell.execute_reply.started": "2022-01-31T06:52:38.995062Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "38c8977ee8374205983750031f5fbe56",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|Count_total|\n",
      "+-----------+\n",
      "|    2000020|\n",
      "+-----------+\n",
      "\n",
      "+--------+\n",
      "|Count_NJ|\n",
      "+--------+\n",
      "|  200000|\n",
      "+--------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT count(*) as Count_total FROM temp_trip_table_v2\").show()\n",
    "spark.sql(\"SELECT count(*) as Count_NJ FROM temp_trip_table_v2 WHERE destination = 'New Jersey'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "dbec6b7b-86d7-4371-a3a8-484721222fd5",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-01-31T06:53:25.163562Z",
     "iopub.status.busy": "2022-01-31T06:53:25.163331Z",
     "iopub.status.idle": "2022-01-31T06:53:30.467011Z",
     "shell.execute_reply": "2022-01-31T06:53:30.466397Z",
     "shell.execute_reply.started": "2022-01-31T06:53:25.163537Z"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "b1f25eb8890c4e33bb45edd185eec30e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|Count_total|\n",
      "+-----------+\n",
      "|    1800020|\n",
      "+-----------+\n",
      "\n",
      "+--------+\n",
      "|Count_NJ|\n",
      "+--------+\n",
      "|       0|\n",
      "+--------+"
     ]
    }
   ],
   "source": [
    "df_v3.createOrReplaceTempView(\"temp_trip_table_v3\")\n",
    "spark.sql(\"SELECT count(*) as Count_total FROM temp_trip_table_v3\").show()\n",
    "spark.sql(\"SELECT count(*) as Count_NJ FROM temp_trip_table_v3 WHERE destination = 'New Jersey'\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c4dfb7b7-fa44-43d7-8e81-96173ed9dbe5",
   "metadata": {},
   "source": [
    "## References\n",
    "\n",
    "- [Apache Spark Configuration - Runtime Environment](https://spark.apache.org/docs/latest/configuration.html#runtime-environment)\n",
    "- [`io.delta` Maven Repository](https://mvnrepository.com/artifact/io.delta)\n",
    "  - [delta-core](https://mvnrepository.com/artifact/io.delta/delta-core)\n",
    "  - [delta-spark](https://mvnrepository.com/artifact/io.delta/delta-spark)\n",
    "- [Use a Delta Lake cluster with Spark in Amazon EMR](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/Deltausing-cluster-spark.html)\n",
    "- [Amazon EMR Release Guide](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-release-components.html)\n",
    "  - [Application versions in Amazon EMR 7.x releases](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-release-app-versions-7.x.html)\n",
    "  - [Application versions in Amazon EMR 6.x releases](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-release-app-versions-6.x.html)"
   ]
  }
 ],
 "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
}