quickstart.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "79f6bcb2-baac-49aa-ba3b-f3bac4e0f9f5",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Delta Lake PySpark Quickstart\n",
    "\n",
    "Referring to https://docs.delta.io/latest/quick-start.html, the following steps have been taken care of by the docker image.\n",
    "\n",
    "> Note: You do **NOT** need to run these 2 commands.\n",
    "\n",
    "## Python Notes\n",
    "```bash\n",
    "pip install pyspark==<compatible-spark-version>\n",
    "\n",
    "$SPARK_HOME/bin/pyspark --packages io.delta:<compatible-delta-version> --conf \"spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension\" --conf \"spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog\"\n",
    "```\n",
    "\n",
    "## Scala Notes\n",
    "If you would like to follow the scala version open a terminal and follow the scala instructions starting with\n",
    "\n",
    "```bash\n",
    "$SPARK_HOME/bin/spark-shell --packages io.delta:<compatible-delta-version> --conf \"spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension\" --conf \"spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog\"\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cff00399-b928-4f43-b8a9-65aae65c9320",
   "metadata": {},
   "source": [
    "## Display versions of `python` and `spark`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "3fdc911f-7361-45c5-b849-43af6eb1652a",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.9.2 (default, Feb 28 2021, 17:03:44) \n",
      "[GCC 10.2.1 20210110]\n"
     ]
    }
   ],
   "source": [
    "import sys\n",
    "print (sys.version)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "3103d8da-796f-4927-b80d-a337ef1fba6b",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'3.3.2'"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e27326eb-b153-455b-aac7-e1fef5b265c6",
   "metadata": {},
   "source": [
    "## Write to and read from a Delta Lake table"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d303e34b-0ebf-4f24-b1cb-be39cea040b9",
   "metadata": {},
   "source": [
    "### Write a Spark DataFrame to a Delta Lake table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "df98ff82-9a19-4744-94ca-0d1e2050c7e4",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/04/26 01:30:13 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "data = spark.range(0, 5)\n",
    "\n",
    "(data\n",
    "  .write\n",
    "  .format(\"delta\")\n",
    "  .save(\"/tmp/delta-table\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8aae7e09-295c-4032-b4b5-8fca0f90f334",
   "metadata": {},
   "source": [
    "### Read the above Delta Lake table to a Spark DataFrame and display the DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "e74258ca-e9fd-4d6a-81da-7100a1836c3d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  0|\n",
      "|  1|\n",
      "|  2|\n",
      "|  3|\n",
      "|  4|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = (spark\n",
    "        .read\n",
    "        .format(\"delta\")\n",
    "        .load(\"/tmp/delta-table\")\n",
    "        .orderBy(\"id\")\n",
    "      )\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "64c3a41b-b2bf-4754-b3d9-998fb00379db",
   "metadata": {},
   "source": [
    "## Overwrite a Delta Lake table"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2a1104e6-98e8-4401-b7c7-007de71b6f91",
   "metadata": {},
   "source": [
    "### Overwrite the Delta Lake table written in the above step"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "91c5e69b-f615-4fd8-97c0-c4e1739740fc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "data = spark.range(5, 10)\n",
    "\n",
    "(data\n",
    "  .write\n",
    "  .format(\"delta\")\n",
    "  .mode(\"overwrite\")\n",
    "  .save(\"/tmp/delta-table\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "11e5297c-f6b4-41e0-8014-54d6958ea352",
   "metadata": {},
   "source": [
    "### Read the above overwritten Delta Lake table to a Spark DataFrame and display the DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ca38e5f9-077b-4fb8-8d3f-74d6c12b554b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  5|\n",
      "|  6|\n",
      "|  7|\n",
      "|  8|\n",
      "|  9|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = (spark\n",
    "        .read\n",
    "        .format(\"delta\")\n",
    "        .load(\"/tmp/delta-table\")\n",
    "        .orderBy(\"id\")\n",
    "      )\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73b416c8-6020-47ac-a017-3a8858a74d8b",
   "metadata": {},
   "source": [
    "## Delta Lake and [ACID](https://en.wikipedia.org/wiki/ACID)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed33f55e-cfa2-4a09-a95d-0238b8f5d9e0",
   "metadata": {},
   "source": [
    "### Showcase `update` feature of Delta Lake and display the resulting DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "a1fec5f6-1aaa-4bc3-a2fa-b357e6bb6c98",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  5|\n",
      "|  7|\n",
      "|  9|\n",
      "|106|\n",
      "|108|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from delta.tables import *\n",
    "from pyspark.sql.functions import *\n",
    "\n",
    "delta_table = DeltaTable.forPath(spark, \"/tmp/delta-table\")\n",
    "\n",
    "# Update every even value by adding 100 to it\n",
    "(delta_table\n",
    "  .update(\n",
    "    condition = expr(\"id % 2 == 0\"),\n",
    "    set = { \"id\": expr(\"id + 100\") }\n",
    "  )\n",
    ")\n",
    "\n",
    "(delta_table\n",
    "  .toDF()\n",
    "  .orderBy(\"id\")\n",
    "  .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "06375c77-855d-4230-8d5f-4a13feabd711",
   "metadata": {},
   "source": [
    "### Showcase `delete` feature of Delta Lake and display the resulting DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "b69423a2-b8b5-4e77-8d7c-492571e3f6c5",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  5|\n",
      "|  7|\n",
      "|  9|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Delete every even value\n",
    "(delta_table\n",
    "  .delete(\n",
    "    condition = expr(\"id % 2 == 0\")\n",
    "  )\n",
    ")\n",
    "\n",
    "(delta_table\n",
    "  .toDF()\n",
    "  .orderBy(\"id\")\n",
    "  .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "172aed35-319a-4a2c-95ce-91d3c4ed1bfd",
   "metadata": {},
   "source": [
    "### Showcase `merge` feature of Delta Lake and display the resulting DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "4dd08581-4d08-4c1a-8cc1-12d0b2ca18d2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/04/26 01:30:27 WARN HintErrorLogger: Hint (strategy=broadcast) is not supported in the query: build left for full outer join.\n",
      "23/04/26 01:30:27 WARN HintErrorLogger: Hint (strategy=broadcast) is not supported in the query: build left for full outer join.\n",
      "23/04/26 01:30:27 WARN HintErrorLogger: Hint (strategy=broadcast) is not supported in the query: build left for full outer join.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  0|\n",
      "|  1|\n",
      "|  2|\n",
      "|  3|\n",
      "|  4|\n",
      "|  5|\n",
      "|  6|\n",
      "|  7|\n",
      "|  8|\n",
      "|  9|\n",
      "| 10|\n",
      "| 11|\n",
      "| 12|\n",
      "| 13|\n",
      "| 14|\n",
      "| 15|\n",
      "| 16|\n",
      "| 17|\n",
      "| 18|\n",
      "| 19|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Upsert (merge) new data\n",
    "new_data = spark.range(0, 20)\n",
    "\n",
    "(delta_table.alias(\"old_data\")\n",
    "  .merge(\n",
    "      new_data.alias(\"new_data\"),\n",
    "      \"old_data.id = new_data.id\"\n",
    "      )\n",
    "  .whenMatchedUpdate(set = { \"id\": col(\"new_data.id\") })\n",
    "  .whenNotMatchedInsert(values = { \"id\": col(\"new_data.id\") })\n",
    "  .execute()\n",
    ")\n",
    "\n",
    "(delta_table\n",
    "  .toDF()\n",
    "  .orderBy(\"id\")\n",
    "  .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b1e437d-93ab-4ced-839a-0f01bf568817",
   "metadata": {},
   "source": [
    "## Time travel feature of Delta Lake"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d40cfbc6-fc4a-4f1f-9b2b-790d7170ba5c",
   "metadata": {},
   "source": [
    "### Display the entire history of the above Delta Lake table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "0c488aac-6eab-40cc-ae66-fb50b6e79e6f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+---------+--------------------+--------------------+--------------------+\n",
      "|version|           timestamp|operation| operationParameters|    operationMetrics|          engineInfo|\n",
      "+-------+--------------------+---------+--------------------+--------------------+--------------------+\n",
      "|      4|2023-04-26 01:30:...|    MERGE|{predicate -> (ol...|{numTargetRowsCop...|Apache-Spark/3.3....|\n",
      "|      3|2023-04-26 01:30:...|   DELETE|{predicate -> [\"(...|{numRemovedFiles ...|Apache-Spark/3.3....|\n",
      "|      2|2023-04-26 01:30:...|   UPDATE|{predicate -> ((i...|{numRemovedFiles ...|Apache-Spark/3.3....|\n",
      "|      1|2023-04-26 01:30:...|    WRITE|{mode -> Overwrit...|{numFiles -> 4, n...|Apache-Spark/3.3....|\n",
      "|      0|2023-04-26 01:30:...|    WRITE|{mode -> ErrorIfE...|{numFiles -> 4, n...|Apache-Spark/3.3....|\n",
      "+-------+--------------------+---------+--------------------+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# get the full history of the table\n",
    "delta_table_history = (DeltaTable\n",
    "                        .forPath(spark, \"/tmp/delta-table\")\n",
    "                        .history()\n",
    "                      )\n",
    "\n",
    "(delta_table_history\n",
    "   .select(\"version\", \"timestamp\", \"operation\", \"operationParameters\", \"operationMetrics\", \"engineInfo\")\n",
    "   .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4aa63ae4-11d0-4aef-9bdb-58fa76c1fd54",
   "metadata": {},
   "source": [
    "### Latest version of the Delta Lake table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "8cbfda73-d20a-4104-9178-040e079a92dc",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+---------+--------------------+--------------------+--------------------+\n",
      "|version|           timestamp|operation| operationParameters|    operationMetrics|          engineInfo|\n",
      "+-------+--------------------+---------+--------------------+--------------------+--------------------+\n",
      "|      4|2023-04-13 06:11:...|    MERGE|{predicate -> (ol...|{numTargetRowsCop...|Apache-Spark/3.3....|\n",
      "|      3|2023-04-13 06:11:...|   DELETE|{predicate -> [\"(...|{numRemovedFiles ...|Apache-Spark/3.3....|\n",
      "|      2|2023-04-13 06:11:...|   UPDATE|{predicate -> ((i...|{numRemovedFiles ...|Apache-Spark/3.3....|\n",
      "|      1|2023-04-13 06:11:...|    WRITE|{mode -> Overwrit...|{numFiles -> 4, n...|Apache-Spark/3.3....|\n",
      "|      0|2023-04-13 06:11:...|    WRITE|{mode -> ErrorIfE...|{numFiles -> 4, n...|Apache-Spark/3.3....|\n",
      "+-------+--------------------+---------+--------------------+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# get the full history of the table\n",
    "delta_table_history = (DeltaTable\n",
    "                        .forPath(spark, \"/tmp/delta-table\")\n",
    "                        .history()\n",
    "                      )\n",
    "\n",
    "(delta_table_history\n",
    "   .select(\"version\", \"timestamp\", \"operation\", \"operationParameters\", \"operationMetrics\", \"engineInfo\")\n",
    "   .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4aa63ae4-11d0-4aef-9bdb-58fa76c1fd54",
   "metadata": {},
   "source": [
    "### Latest version of the Delta Lake table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "8cbfda73-d20a-4104-9178-040e079a92dc",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  0|\n",
      "|  1|\n",
      "|  2|\n",
      "|  3|\n",
      "|  4|\n",
      "|  5|\n",
      "|  6|\n",
      "|  7|\n",
      "|  8|\n",
      "|  9|\n",
      "| 10|\n",
      "| 11|\n",
      "| 12|\n",
      "| 13|\n",
      "| 14|\n",
      "| 15|\n",
      "| 16|\n",
      "| 17|\n",
      "| 18|\n",
      "| 19|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = (spark\n",
    "        .read\n",
    "        .format(\"delta\")\n",
    "        .load(\"/tmp/delta-table\")\n",
    "        .orderBy(\"id\")\n",
    "      )\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e7cce27-2852-4fc8-b61c-747023f90658",
   "metadata": {},
   "source": [
    "### Time travel to the version `0` of the Delta Lake table using Delta Lake's history feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "dbb43006-112b-4841-80fd-6179c802d8ec",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  0|\n",
      "|  1|\n",
      "|  2|\n",
      "|  3|\n",
      "|  4|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = (spark\n",
    "        .read\n",
    "        .format(\"delta\")\n",
    "        .option(\"versionAsOf\", 0) # we pass an option `versionAsOf` with the required version number we are interested in\n",
    "        .load(\"/tmp/delta-table\")\n",
    "        .orderBy(\"id\")\n",
    "      )\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d77368f4-2d51-493b-8c48-362b0ceedc38",
   "metadata": {},
   "source": [
    "### Time travel to the version `3` of the Delta Lake table using Delta Lake's  history feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "9daa8958-237b-4be2-b397-759ea9686b63",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  5|\n",
      "|  7|\n",
      "|  9|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = (spark\n",
    "        .read\n",
    "        .format(\"delta\")\n",
    "        .option(\"versionAsOf\", 3) # we pass an option `versionAsOf` with the required version number we are interested in\n",
    "        .load(\"/tmp/delta-table\")\n",
    "        .orderBy(\"id\")\n",
    "      )\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fd078e9b-9682-4027-aa63-96ae1c32b2eb",
   "metadata": {},
   "source": [
    "## A little bit of Streaming"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "e0840907-f972-43de-9233-59076f32acc7",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/04/26 01:31:34 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "streaming_df = (spark\n",
    "                 .readStream\n",
    "                 .format(\"rate\")\n",
    "                 .load()\n",
    "               )\n",
    "\n",
    "stream = (streaming_df\n",
    "            .selectExpr(\"value as id\")\n",
    "            .writeStream\n",
    "            .format(\"delta\")\n",
    "            .option(\"checkpointLocation\", \"/tmp/checkpoint\")\n",
    "            .start(\"/tmp/delta-table\")\n",
    "          )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "769b4446-11f8-4948-9f76-cd709ba65bd5",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/04/26 01:31:41 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-54eea548-8c64-4060-8d84-0a904a29eebc. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n",
      "23/04/26 01:31:41 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  0|\n",
      "|  1|\n",
      "|  2|\n",
      "|  3|\n",
      "|  4|\n",
      "|  5|\n",
      "|  6|\n",
      "|  7|\n",
      "|  8|\n",
      "|  9|\n",
      "| 10|\n",
      "| 11|\n",
      "| 12|\n",
      "| 13|\n",
      "| 14|\n",
      "| 15|\n",
      "| 16|\n",
      "| 17|\n",
      "| 18|\n",
      "| 19|\n",
      "+---+\n",
      "only showing top 20 rows\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  6|\n",
      "|  7|\n",
      "+---+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 2\n",
      "-------------------------------------------\n",
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "|  8|\n",
      "|  9|\n",
      "| 10|\n",
      "+---+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 3\n",
      "-------------------------------------------\n",
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "| 11|\n",
      "+---+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 4\n",
      "-------------------------------------------\n",
      "+---+\n",
      "| id|\n",
      "+---+\n",
      "| 12|\n",
      "| 13|\n",
      "+---+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 172:==================================================>    (46 + 4) / 50]\r"
     ]
    }
   ],
   "source": [
    "# To view the results of this step, view your container logs after execution using: docker logs --follow <first 4 number of container id>\n",
    "\n",
    "stream2 = (spark\n",
    "            .readStream\n",
    "            .format(\"delta\")\n",
    "            .load(\"/tmp/delta-table\")\n",
    "            .writeStream\n",
    "            .format(\"console\")\n",
    "            .start()\n",
    "          )"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.2"
  },
  "vscode": {
   "interpreter": {
    "hash": "916dbcbb3f70747c44a77c7bcd40155683ae19c65e1c03b4aa3499c5328201f1"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}